Repository: Netflix/consoleme
File: consoleme/lib/oidc.py (Python)

Code before:
            "jwks_uri": jwks_uri,
        }
    client_id = config.get("oidc_secrets.client_id")
    client_secret = config.get("oidc_secrets.secret")
    if not (client_id or client_secret):
        raise MissingConfigurationValue("Missing OIDC Secrets")
    oidc_config["client_id"] = client_id
    oidc_config["client_secret"] = client_secret
    oidc_config["jwt_keys"] = {}
    jwks_uris = [oidc_config["jwks_uri"]]
    jwks_uris.extend(config.get("get_user_by_oidc_settings.extra_jwks_uri", []))
    for jwks_uri in jwks_uris:
        # Fetch jwks_uri for jwt validation
        res = await http_client.fetch(
            jwks_uri,
            method="GET",
            headers={
                "Content-Type": "application/x-www-form-urlencoded",
                "Accept": "application/json",
            },
        )
        jwks_data = json.loads(res.body)
        for k in jwks_data["keys"]:
            key_type = k["kty"]
            key_id = k["kid"]
            if key_type == "RSA":
                oidc_config["jwt_keys"][key_id] = RSAAlgorithm.from_jwk(json.dumps(k))
            elif key_type == "EC":
                oidc_config["jwt_keys"][key_id] = ECAlgorithm.from_jwk(json.dumps(k))
    return oidc_config


async def authenticate_user_by_oidc(request):
    email = None
    groups = []
    decoded_access_token = {}
    oidc_config = await populate_oidc_config()
    function = f"{__name__}.{sys._getframe().f_code.co_name}"
    log_data = {"function": function}
    code = request.get_argument("code", None)
    protocol = request.request.protocol
    if "https://" in config.get("url"):
        # If we're behind a load balancer that terminates tls for us, request.request.protocol will be "http://" and our
        # oidc redirect will be invalid
        protocol = "https"
    force_redirect = await should_force_redirect(request.request)

    # The endpoint where we want our OIDC provider to redirect us back to perform auth
    oidc_redirect_uri = f"{protocol}://{request.request.host}/auth"

    # The endpoint where the user wants to be sent after authentication. This will be stored in the state

Review comment (security):
Do we need to protect against `key_type = "none"` here? This isn't in the authn flow, but that might help prevent insecure configurations. Alternatively, we could slap on an `else` that raises an exception.

Code after:
            "jwks_uri": jwks_uri,
        }
    client_id = config.get("oidc_secrets.client_id")
    client_secret = config.get("oidc_secrets.secret")
    if not (client_id or client_secret):
        raise MissingConfigurationValue("Missing OIDC Secrets")
    oidc_config["client_id"] = client_id
    oidc_config["client_secret"] = client_secret
    oidc_config["jwt_keys"] = {}
    jwks_uris = [oidc_config["jwks_uri"]]
    jwks_uris.extend(config.get("get_user_by_oidc_settings.extra_jwks_uri", []))
    for jwks_uri in jwks_uris:
        # Fetch jwks_uri for jwt validation
        res = await http_client.fetch(
            jwks_uri,
            method="GET",
            headers={
                "Content-Type": "application/x-www-form-urlencoded",
                "Accept": "application/json",
            },
        )
        jwks_data = json.loads(res.body)
        for k in jwks_data["keys"]:
            key_type = k["kty"]
            key_id = k["kid"]
            if key_type == "RSA":
                oidc_config["jwt_keys"][key_id] = RSAAlgorithm.from_jwk(json.dumps(k))
            elif key_type == "EC":
                oidc_config["jwt_keys"][key_id] = ECAlgorithm.from_jwk(json.dumps(k))
            else:
                raise MissingConfigurationValue(
                    f"OIDC/OAuth2 key type not recognized. Detected key type: {key_type}."
                )
    return oidc_config


async def authenticate_user_by_oidc(request):
    email = None
    groups = []
    decoded_access_token = {}
    oidc_config = await populate_oidc_config()
    function = f"{__name__}.{sys._getframe().f_code.co_name}"
    log_data = {"function": function}
    code = request.get_argument("code", None)
    protocol = request.request.protocol
    if "https://" in config.get("url"):
        # If we're behind a load balancer that terminates tls for us, request.request.protocol will be "http://" and our
        # oidc redirect will be invalid
        protocol = "https"
    force_redirect = await should_force_redirect(request.request)