fxa/oauth.py

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import json

import os
import base64
import hashlib
from urllib.parse import urlparse, urlunparse, urlencode, parse_qs

import jwt
from fxa.cache import MemoryCache, DEFAULT_CACHE_EXPIRY
from fxa.constants import PRODUCTION_URLS
from fxa.errors import OutOfProtocolError, ScopeMismatchError, TrustError
from fxa._utils import APIClient, scope_matches, get_hmac, HawkTokenAuth

DEFAULT_SERVER_URL = PRODUCTION_URLS['oauth']
VERSION_SUFFIXES = ("/v1",)
TOKEN_HMAC_SECRET = 'PyFxA Token Cache Hmac Secret'


class Client:
    """Client for talking to the Firefox Accounts OAuth server"""

    def __init__(self, client_id=None, client_secret=None, server_url=None,
                 cache=True, ttl=DEFAULT_CACHE_EXPIRY, jwks=None):
        self.client_id = client_id
        self.client_secret = client_secret
        if server_url is None:
            server_url = DEFAULT_SERVER_URL
        server_url = server_url.rstrip('/')
        if not server_url.endswith(VERSION_SUFFIXES):
            server_url += VERSION_SUFFIXES[0]
        if isinstance(server_url, str):
            self.apiclient = APIClient(server_url)
        else:
            self.apiclient = server_url

        self.cache = cache
        if self.cache is True:
            self.cache = MemoryCache(ttl)

        if jwks is not None:
            # Fail early if bad JWKs were provided.
            for key in jwks:
                jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(key))
        self.jwks = jwks

    @property
    def server_url(self):
        return self.apiclient.server_url

    def get_client_metadata(self, client_id=None):
        """Get the OAuth client metadata for a given client_id."""
        if client_id is None:
            client_id = self.client_id
        return self.apiclient.get(f"/client/{client_id}")

    def get_redirect_url(self, state="", redirect_uri=None, scope=None,
                         action=None, email=None, client_id=None,
                         code_challenge=None, code_challenge_method=None,
                         access_type=None, keys_jwk=None):
        """Get the URL to redirect to to initiate the oauth flow."""
        if client_id is None:
            client_id = self.client_id
        params = {
            "client_id": client_id,
            "state": state,
        }
        if redirect_uri is not None:
            params["redirect_uri"] = redirect_uri
        if scope is not None:
            params["scope"] = scope
        if action is not None:
            params["action"] = action
        if email is not None:
            params["email"] = email
        if code_challenge is not None:
            params["code_challenge"] = code_challenge
        if code_challenge_method is not None:
            params["code_challenge_method"] = code_challenge_method
        if keys_jwk is not None:
            params["keys_jwk"] = keys_jwk
        if access_type is not None:
            params["access_type"] = access_type
        query_str = urlencode(params)
        authorization_url = urlparse(self.server_url + "/authorization")
        return urlunparse(authorization_url._replace(query=query_str))

    def trade_code(self, code, client_id=None, client_secret=None,
                   code_verifier=None, ttl=None):
        """Trade the authentication code for a longer lived token.

        :param code: the authentication code from the oauth redirect dance.
        :param client_id: the string generated during FxA client registration.
        :param client_secret: the related secret string.
        :param code_verifier: optional PKCE code verifier.
        :param ttl: optional ttl in seconds, the access token is valid for.
        :returns: a dict with user id and authorized scopes for this token.
        """
        if client_id is None:
            client_id = self.client_id
        if client_secret is None:
            client_secret = self.client_secret
        url = '/token'
        body = {
            'code': code,
            'client_id': client_id,
        }
        if client_secret is not None:
            body["client_secret"] = client_secret
        if code_verifier is not None:
            body["code_verifier"] = code_verifier
        if ttl is not None:
            body["ttl"] = ttl
        resp = self.apiclient.post(url, body)

        if 'access_token' not in resp:
            error_msg = 'access_token missing in OAuth response'
            raise OutOfProtocolError(error_msg)

        return resp

    def authorize_code(self, session, scope=None, client_id=None,
                       code_challenge=None, code_challenge_method=None):
        """Trade a session for an oauth authorization code.

        This method takes a session for a user and uses it to
        generate an oauth authentication code.  This code can in turn be
        traded for a full-blown oauth token.

        :param session: an auth session to use.
        :param scope: optional scope to be provided by the token.
        :param client_id: the string generated during FxA client registration.
        :param code_challenge: optional PKCE code challenge.
        :param code_challenge_method: optional PKCE code challenge method.
        """
        auth = HawkTokenAuth(session.token, "sessionToken", self.apiclient)

        if client_id is None:
            client_id = self.client_id
        url = "/oauth/authorization"

        # Although not relevant in this scenario from a security perspective,
        # we generate a random 'state' and check the returned redirect URL
        # for completeness.
        state = base64.urlsafe_b64encode(os.urandom(24)).decode('utf-8')

        body = {
            "client_id": client_id,
            "state": state
        }
        if scope is not None:
            body["scope"] = scope
        if code_challenge is not None:
            body["code_challenge"] = code_challenge
            body["code_challenge_method"] = code_challenge_method or "S256"
        resp = self.apiclient.post(url, body, auth=auth)

        if "redirect" not in resp:
            error_msg = "redirect missing in OAuth response"
            raise OutOfProtocolError(error_msg)

        # This flow is designed for web-based redirects.
        # In order to get the code we must parse it from the redirect url.
        query_params = parse_qs(urlparse(resp["redirect"]).query)

        # Check that the 'state' parameter is present and the same we provided
        if "state" not in query_params:
            error_msg = "state missing in OAuth response"
            raise OutOfProtocolError(error_msg)

        if state != query_params["state"][0]:
            error_msg = f"state mismatch in OAuth response (wanted: '{state}', got: '{query_params['state'][0]}')"
            raise OutOfProtocolError(error_msg)

        try:
            return query_params["code"][0]
        except (KeyError, IndexError, ValueError):
            error_msg = "code missing in OAuth redirect url"
            raise OutOfProtocolError(error_msg)

    def authorize_token(self, session, scope=None, client_id=None):
        """Trade session token for an oauth token.

        :param session: an auth session.
        :param scope: optional scope to be provided by the token.
        :param client_id: the string generated during FxA client registration.
        """
        (challenge, verifier) = self.generate_pkce_challenge()
        code = self.authorize_code(session, scope, client_id, **challenge)
        resp = self.trade_code(code, **verifier)

        if 'access_token' not in resp:
            error_msg = 'access_token missing in OAuth response'
            raise OutOfProtocolError(error_msg)

        return resp['access_token']

    def _verify_jwt_token(self, key, token):
        pubkey = jwt.algorithms.RSAAlgorithm.from_jwk(key)
        # The FxA OAuth ecosystem currently doesn't make good use of aud, and
        # instead relies on scope for restricting which services can accept
        # which tokens. So there's no value in checking it here, and in fact if
        # we check it here, it fails because the right audience isn't being
        # requested.
        decoded = jwt.decode(
            token, pubkey, algorithms=['RS256'], options={'verify_aud': False}
        )
        # Ref https://tools.ietf.org/html/rfc7515#section-4.1.9 the `typ` header
        # is lowercase and has an implicit default `application/` prefix.
        typ = jwt.get_unverified_header(token).get('typ', '')
        if '/' not in typ:
            typ = 'application/' + typ
        if typ.lower() != 'application/at+jwt':
            raise TrustError
        return {
            'user': decoded.get('sub'),
            'client_id': decoded.get('client_id'),
            'scope': decoded.get('scope', '').split(),
            'generation': decoded.get('fxa-generation'),
            'profile_changed_at': decoded.get('fxa-profileChangedAt')
        }

    def verify_token(self, token, scope=None):
        """Verify an OAuth token, and retrieve user id and scopes.

        :param token: the string to verify.
        :param scope: optional scope expected to be provided for this token.
        :returns: a dict with user id and authorized scopes for this token.
        :raises fxa.errors.ClientError: if the provided token is invalid.
        :raises fxa.errors.TrustError: if the token scopes do not match.
        """
        key = 'fxa.oauth.verify_token:%s:%s' % (
            get_hmac(token, TOKEN_HMAC_SECRET), scope)
        if self.cache is not None:
            resp = self.cache.get(key)
        else:
            resp = None

        if resp is None:
            # We want to fetch
            # https://oauth.accounts.firefox.com/.well-known/openid-configuration
            # and then get the jwks_uri key to get the /jwks url, but we'll
            # just hardcodes it like this for now; our /jwks url will never
            # change.
            # https://github.com/mozilla/PyFxA/issues/81 is an issue about
            # getting the jwks url out of the openid-configuration.
            keys = []
            if self.jwks is not None:
                keys.extend(self.jwks)
            else:
                keys.extend(self.apiclient.get('/jwks').get('keys', []))
            resp = None
            try:
                for k in keys:
                    try:
                        resp = self._verify_jwt_token(json.dumps(k), token)
                        break
                    except jwt.exceptions.InvalidSignatureError:
                        # It's only worth trying other keys in the event of
                        # `InvalidSignature`; if it was invalid for other reasons
                        # (e.g. it's expired) then using a different key won't
                        # help.
                        continue
                else:
                    # It's a well-formed JWT, but not signed by any of the advertized keys.
                    # We can immediately surface this as an error.
                    if len(keys) > 0:
                        raise TrustError({"error": "invalid signature"})
            except (jwt.exceptions.DecodeError, jwt.exceptions.InvalidKeyError):
                # It wasn't a JWT at all, or it was signed using a key type we
                # don't support. Fall back to asking the FxA server to verify.
                pass
            except jwt.exceptions.PyJWTError as e:
                # Any other JWT-related failure (e.g. expired token) can
                # immediately surface as a trust error.
                raise TrustError({"error": str(e)})
            if resp is None:
                resp = self.apiclient.post('/verify', {'token': token})
            missing_attrs = ", ".join([
                k for k in ('user', 'scope', 'client_id')
                if resp.get(k) is None
            ])
            if missing_attrs:
                error_msg = f'{missing_attrs} missing in OAuth response'
                raise OutOfProtocolError(error_msg)

            if scope is not None:
                authorized_scope = resp['scope']
                if not scope_matches(authorized_scope, scope):
                    raise ScopeMismatchError(authorized_scope, scope)

            if self.cache is not None:
                self.cache.set(key, json.dumps(resp))
        else:
            resp = json.loads(resp)

        return resp

    def destroy_token(self, token):
        """Destroy an OAuth token

        :param token: the token to destroy.
        :raises fxa.errors.ClientError: if the provided token is invalid.
        """
        url = '/destroy'
        body = {
            'token': token
        }
        self.apiclient.post(url, body)

    def generate_pkce_challenge(self):
        """Ramdomly generate parameters for a PKCE challenge.

        This method returns a two-tuple (challenge, response) where the first
        item contains request parameters for a PKCE challenge, and the second
        item contains the corresponding parameters for a verification.
        """
        code_verifier = base64.urlsafe_b64encode(os.urandom(32)).decode('utf-8').rstrip("=")
        raw_challenge = hashlib.sha256(code_verifier.encode('utf-8')).digest()
        code_challenge = base64.urlsafe_b64encode(raw_challenge).decode('utf-8').rstrip("=")
        return ({
            "code_challenge": code_challenge,
            "code_challenge_method": "S256",
        }, {
            "code_verifier": code_verifier,
        })