import logging from urllib.parse import urlsplit, urlunsplit from django.conf import settings from djagger.decorators import schema from django.http import Http404 from django.http import HttpResponse from django.http import JsonResponse from spid_cie_oidc.entity.jwtse import create_jws from spid_cie_oidc.entity.schemas.resolve_endpoint import ( ResolveRequest, ResolveResponse, ResolveErrorResponse ) from spid_cie_oidc.entity.trust_chain_operations import get_or_create_trust_chain from .models import ( FederationEntityConfiguration, FederationHistoricalKey, TrustChain, StaffToken ) from .settings import HTTPC_PARAMS from .statements import OIDCFED_FEDERATION_WELLKNOWN_URL from .utils import iat_now logger = logging.getLogger(__name__) def _swap_scheme(url: str) -> str: parts = urlsplit(url) if parts.scheme == "http": return urlunsplit(("https", parts.netloc, parts.path, parts.query, parts.fragment)) if parts.scheme == "https": return urlunsplit(("http", parts.netloc, parts.path, parts.query, parts.fragment)) return url def get_subs_from_wellknown(request, wkuri :str): sub = request.build_absolute_uri().split(wkuri)[0].rstrip("/") sub_values = [] for candidate in (sub, _swap_scheme(sub)): for value in (candidate, f"{candidate}/"): if value not in sub_values: sub_values.append(value) return sub_values def entity_configuration(request): """ OIDC Federation Entity Configuration at .well-known/openid-federation """ _sub_values = get_subs_from_wellknown(request, OIDCFED_FEDERATION_WELLKNOWN_URL) conf = FederationEntityConfiguration.objects.filter( # TODO: check for reverse proxy and forwarders ... sub__in=_sub_values, is_active=True, ).first() if not conf: conf = FederationEntityConfiguration.objects.filter(is_active=True).first() if not conf: # pragma: no cover raise Http404() if request.GET.get("format") == "json": # pragma: no cover return JsonResponse(conf.entity_configuration_as_dict, safe=False) else: return HttpResponse( conf.entity_configuration_as_jws, content_type="application/entity-statement+jwt" ) @schema( methods=['GET'], get_request_schema = { "application/x-www-form-urlencoded": ResolveRequest }, get_response_schema = { "400": ResolveErrorResponse, "404": ResolveErrorResponse, "200": ResolveResponse }, tags = ['Federation API'] ) def resolve_entity_statement(request, format: str = "jose"): """ resolves the final metadata of its descendants In this implementation we only returns a preexisting Metadata if it's valid we avoid any possibility to trigger a new Metadata discovery if """ if not all((request.GET.get("sub"), request.GET.get("trust_anchor"))): return JsonResponse( { "error": "invalid_request", "error_description": "sub and trust_anchor parameters are REQUIRED." }, status=400 ) iss = FederationEntityConfiguration.objects.filter(is_active=True).first() _q = dict( sub=request.GET["sub"], trust_anchor__sub=request.GET["trust_anchor"], is_active=True ) # gets the cached one entity = TrustChain.objects.filter(**_q).first() # only with privileged actors with staff token can triggers a new trust chain staff_token_head = request.headers.get("Authorization", None) if staff_token_head: staff_token = StaffToken.objects.filter( token = staff_token_head ).first() if staff_token.is_valid: try: # a staff token get a fresh trust chain on each call entity = get_or_create_trust_chain( httpc_params = HTTPC_PARAMS, required_trust_marks = getattr( settings, "OIDCFED_REQUIRED_TRUST_MARKS", [] ), subject=_q["sub"], trust_anchor=_q["trust_anchor__sub"], force = True ) except Exception as e: logger.error( f"Failed privileged Trust Chain creation for {_q['sub']}: {e}" ) if not entity: return JsonResponse( { "error": "not_found", "error_description": "entity not found" }, status=404 ) metadata = entity.metadata # Filter by entity_type if requested (OpenID Federation 8.3.1) entity_types = request.GET.getlist("entity_type") if entity_types: metadata = { k: v for k, v in (metadata or {}).items() if k in entity_types } res = { "iss": iss.sub, "sub": request.GET["sub"], "iat": entity.iat_as_timestamp, "exp": entity.exp_as_timestamp, "trust_marks": entity.trust_marks, "metadata": metadata, "trust_chain": entity.chain } if request.GET.get("format") == "json" or format == "json": return JsonResponse(res, safe=False) jwk = iss.jwks_fed[0] signed = create_jws( res, jwk, alg=iss.default_signature_alg, protected={"typ": "resolve-response+jwt", "kid": jwk.get("kid")}, ) return HttpResponse( signed, content_type="application/resolve-response+jwt", ) def openid_jwks(request, metadata_type:str, resource_type:str): """ resource_tytpe = set(jwks_uri, jwks.jose) """ _sub = request.build_absolute_uri().rsplit(resource_type)[0] _lookup = _sub.replace(f"/{metadata_type}/", "").rstrip("/") _lookup_values = [_lookup, _swap_scheme(_lookup)] conf = FederationEntityConfiguration.objects.filter( # TODO: check for reverse proxy and forwarders ... sub__in=_lookup_values, is_active=True, ).first() if not conf: conf = FederationEntityConfiguration.objects.filter(is_active=True).first() if not conf: # pragma: no cover raise Http404() content = conf.entity_configuration_as_dict['metadata'].get( metadata_type, {} ).get("jwks", {}) if not content: raise Http404() if resource_type == 'jwks.json': return JsonResponse(content, safe=False) elif resource_type == 'jwks.jose': return HttpResponse( create_jws(content, conf.jwks_fed[0]), content_type="application/jose" ) else: raise Http404() def openid_connect_jwks_uri(request, metadata_type, resource_type="jwks.json"): """ OpenID Connect default jwks_uri """ return openid_jwks(request, metadata_type, resource_type) def openid_connect_signed_jwks_uri(request, metadata_type, resource_type="jwks.jose"): """ OpenID Federation signed_jwks_uri """ return openid_jwks(request, metadata_type, resource_type) def historical_keys(request): """ OIDC Federation Entity Configuration at .well-known/openid-federation """ sub_values = get_subs_from_wellknown(request, '.well-known/openid-federation-historical-jwks') keys = FederationHistoricalKey.objects.filter(entity__sub__in=sub_values) entity = keys.first().entity if keys.first() else None reg = { "iss": entity.sub if entity else sub_values[0], "iat": iat_now(), "keys": [] } for i in keys: reg["keys"].append(i.as_dict) if request.GET.get("format") == "json": # pragma: no cover return JsonResponse(reg, safe=False) else: res = create_jws( reg, entity.jwks_fed[0], typ="jwk-set+jwt", # **kwargs, ) return HttpResponse( res, content_type="application/jwk-set+jwt" )