def authorize_code()

in fxa/oauth.py [0:0]

• 32 lines of code
• 13 McCabe index (conditional complexity)

    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)