Skip to content

castlecraft_engineer.application.auth._jwks_mixin

castlecraft_engineer.application.auth._jwks_mixin

JwksMixin

Bases: AuthenticationServiceBase

Source code in src/castlecraft_engineer/application/auth/_jwks_mixin.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class JwksMixin(AuthenticationServiceBase):
    def _is_jwks_valid(self, jwks_data: dict) -> bool:
        """Validates the structure of JWKS data."""
        if (
            not isinstance(jwks_data, dict)
            or "keys" not in jwks_data
            or not isinstance(jwks_data["keys"], list)
        ):
            return False

        for key in jwks_data["keys"]:
            if not isinstance(key, dict) or "kty" not in key or "kid" not in key:
                return False
        return True

    async def get_active_jwks_response(self) -> Optional[dict]:
        """Fetches JWKS from cache or URL."""
        jwks_url = os.environ.get(ENV_JWKS_URL)
        if not jwks_url:
            self._logger.warning(
                f"{ENV_JWKS_URL} environment variable not set.",
            )
            return None

        async_cache_client = await self._get_resolved_async_cache_client()  # type: ignore
        if async_cache_client:
            jwks_response = await self._get_cached_value_async(JWKS_RESPONSE_KEY)  # type: ignore
        else:
            jwks_response = self._get_cached_value(JWKS_RESPONSE_KEY)  # type: ignore

        if jwks_response:
            if self._is_jwks_valid(jwks_response):
                return jwks_response
            else:
                self._logger.warning(
                    "Cached JWKS data is invalid. Fetching fresh.",
                )
                if async_cache_client:
                    await self._delete_cached_value_async(JWKS_RESPONSE_KEY)  # type: ignore
                else:
                    self._delete_cached_value(JWKS_RESPONSE_KEY)  # type: ignore

        self._logger.info(f"Fetching JWKS from URL: {jwks_url}")
        try:
            response = requests.get(
                jwks_url,
                timeout=10,
                verify=self._request_verify_ssl,
            )
            response.raise_for_status()
            jwks_response = response.json()
        except HTTPError as e:
            self._logger.error(
                f"HTTPError fetching JWKS from {jwks_url}: {e}", exc_info=True
            )
            return None
        except ValueError as e:  # For JSON decoding errors
            self._logger.error(
                f"ValueError decoding JWKS JSON from {jwks_url}: {e}", exc_info=True
            )
            return None

        if not self._is_jwks_valid(jwks_response):
            self._logger.error(f"Fetched JWKS data from {jwks_url} is invalid.")
            return None

        if async_cache_client:
            await self._set_cached_value_async(JWKS_RESPONSE_KEY, jwks_response, ttl=self.JWKS_TTL_SEC)  # type: ignore
        else:
            self._set_cached_value(JWKS_RESPONSE_KEY, jwks_response, ttl=self.JWKS_TTL_SEC)  # type: ignore
        return jwks_response

get_active_jwks_response() async

Fetches JWKS from cache or URL.

Source code in src/castlecraft_engineer/application/auth/_jwks_mixin.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
async def get_active_jwks_response(self) -> Optional[dict]:
    """Fetches JWKS from cache or URL."""
    jwks_url = os.environ.get(ENV_JWKS_URL)
    if not jwks_url:
        self._logger.warning(
            f"{ENV_JWKS_URL} environment variable not set.",
        )
        return None

    async_cache_client = await self._get_resolved_async_cache_client()  # type: ignore
    if async_cache_client:
        jwks_response = await self._get_cached_value_async(JWKS_RESPONSE_KEY)  # type: ignore
    else:
        jwks_response = self._get_cached_value(JWKS_RESPONSE_KEY)  # type: ignore

    if jwks_response:
        if self._is_jwks_valid(jwks_response):
            return jwks_response
        else:
            self._logger.warning(
                "Cached JWKS data is invalid. Fetching fresh.",
            )
            if async_cache_client:
                await self._delete_cached_value_async(JWKS_RESPONSE_KEY)  # type: ignore
            else:
                self._delete_cached_value(JWKS_RESPONSE_KEY)  # type: ignore

    self._logger.info(f"Fetching JWKS from URL: {jwks_url}")
    try:
        response = requests.get(
            jwks_url,
            timeout=10,
            verify=self._request_verify_ssl,
        )
        response.raise_for_status()
        jwks_response = response.json()
    except HTTPError as e:
        self._logger.error(
            f"HTTPError fetching JWKS from {jwks_url}: {e}", exc_info=True
        )
        return None
    except ValueError as e:  # For JSON decoding errors
        self._logger.error(
            f"ValueError decoding JWKS JSON from {jwks_url}: {e}", exc_info=True
        )
        return None

    if not self._is_jwks_valid(jwks_response):
        self._logger.error(f"Fetched JWKS data from {jwks_url} is invalid.")
        return None

    if async_cache_client:
        await self._set_cached_value_async(JWKS_RESPONSE_KEY, jwks_response, ttl=self.JWKS_TTL_SEC)  # type: ignore
    else:
        self._set_cached_value(JWKS_RESPONSE_KEY, jwks_response, ttl=self.JWKS_TTL_SEC)  # type: ignore
    return jwks_response