459 lines
17 KiB
Python
459 lines
17 KiB
Python
import json
|
|
from django.core.paginator import Paginator
|
|
from django.shortcuts import render, redirect
|
|
from django.contrib import messages
|
|
from django.utils.translation import gettext as _
|
|
|
|
from spid_cie_oidc.authority.schemas.fetch_endpoint_request import FetchRequest
|
|
from spid_cie_oidc.authority.schemas.list_endpoint import ListRequest, ListResponse
|
|
from spid_cie_oidc.entity.schemas.resolve_endpoint import ResolveRequest, ResolveResponse
|
|
from spid_cie_oidc.authority.schemas.trust_mark_status_endpoint import TrustMarkRequest, TrustMarkResponse
|
|
from spid_cie_oidc.entity.schemas.fa_metadata import FAMetadata
|
|
|
|
|
|
from .forms import (
|
|
OnboardingRegistrationForm,
|
|
OnboardingCreateTrustChain,
|
|
OnboardingValidatingTrustMarkForm,
|
|
OnboardingDecodeForm
|
|
)
|
|
from .models import OnBoardingRegistration
|
|
from spid_cie_oidc.entity.jwks import (
|
|
private_pem_from_jwk,
|
|
public_pem_from_jwk,
|
|
new_rsa_key,
|
|
serialize_key,
|
|
private_jwk_from_pem,
|
|
public_jwk_from_pem
|
|
)
|
|
|
|
from spid_cie_oidc.entity.jwtse import unpad_jwt_head, unpad_jwt_payload, verify_jws, decrypt_jwe
|
|
from spid_cie_oidc.authority.views import trust_mark_status
|
|
from spid_cie_oidc.authority.validators import validate_entity_configuration
|
|
from spid_cie_oidc.provider.schemas.authn_requests import AuthenticationRequestSpid
|
|
from spid_cie_oidc.provider.schemas.authn_response import AuthenticationResponse
|
|
from spid_cie_oidc.provider.schemas.authn_response import AuthenticationErrorResponse
|
|
from spid_cie_oidc.provider.schemas.introspection_request import IntrospectionRequest
|
|
from spid_cie_oidc.provider.schemas.introspection_response import IntrospectionResponse
|
|
from spid_cie_oidc.provider.schemas.introspection_response import IntrospectionErrorResponseSpid
|
|
from spid_cie_oidc.entity.schemas.op_metadata import OPMetadataSpid, OPMetadataCie
|
|
from spid_cie_oidc.entity.schemas.rp_metadata import RPMetadataSpid,RPMetadataCie
|
|
from spid_cie_oidc.provider.schemas.revocation_request import RevocationRequest
|
|
from spid_cie_oidc.provider.schemas.token_requests import TokenAuthnCodeRequest
|
|
from spid_cie_oidc.provider.schemas.token_response import TokenResponse
|
|
from spid_cie_oidc.provider.schemas.token_response import TokenErrorResponse
|
|
|
|
from spid_cie_oidc.provider.schemas.token_requests import TokenRefreshRequest
|
|
from spid_cie_oidc.provider.schemas.token_response import TokenRefreshResponse
|
|
|
|
from spid_cie_oidc.provider.schemas.jwt import JwtStructure
|
|
from spid_cie_oidc.entity.policy import apply_policy
|
|
|
|
from spid_cie_oidc.relying_party.settings import RP_PROVIDER_PROFILES
|
|
from spid_cie_oidc.provider.settings import OIDCFED_PROVIDER_PROFILES
|
|
from spid_cie_oidc.entity.views import resolve_entity_statement
|
|
|
|
|
|
def onboarding_landing(request): # pragma: no cover
|
|
return render(request, "onboarding_landing.html")
|
|
|
|
|
|
def onboarding_registration(request):
|
|
form = OnboardingRegistrationForm()
|
|
context = {"form": form}
|
|
if request.method == "POST":
|
|
form = OnboardingRegistrationForm(request.POST)
|
|
if not form.is_valid():
|
|
context = {"form": form}
|
|
else:
|
|
form_dict = {**form.cleaned_data}
|
|
OnBoardingRegistration.objects.create(**form_dict)
|
|
messages.success(request, _("Registration successfully"))
|
|
return redirect("oidc_onboarding_entities")
|
|
return render(request, "onboarding_registration.html", context)
|
|
|
|
|
|
def onboarding_entities(request):
|
|
entity_list = OnBoardingRegistration.objects.all()
|
|
p = Paginator(entity_list, 10)
|
|
page = request.GET.get("page")
|
|
entities = p.get_page(page)
|
|
return render(
|
|
request,
|
|
"onboarding_entities.html",
|
|
{"entity_list": entity_list, "entities": entities},
|
|
)
|
|
|
|
|
|
def onboarding_create_jwk(request):
|
|
_rsa_key = new_rsa_key()
|
|
private_jwk = serialize_key(_rsa_key.priv_key, 'private')
|
|
public_jwk = serialize_key(_rsa_key.pub_key)
|
|
context = {
|
|
"private_jwk": json.dumps(private_jwk, indent=4),
|
|
"public_jwk": json.dumps(public_jwk, indent=4)
|
|
}
|
|
return render(request, 'onboarding_jwk.html', context)
|
|
|
|
|
|
def onboarding_convert_jwk(request):
|
|
jwk_type = request.GET.get('type')
|
|
context = {
|
|
"jwk": "",
|
|
"pem": "",
|
|
"jwk_type": jwk_type
|
|
}
|
|
if request.method == 'POST':
|
|
try:
|
|
jwk_str = request.POST.get('jwk')
|
|
jwk_str_double_quote = jwk_str.replace("'", '"')
|
|
jwk_dict = json.loads(jwk_str_double_quote)
|
|
|
|
if jwk_type == 'private':
|
|
pem = private_pem_from_jwk(jwk_dict)
|
|
if jwk_type == 'public':
|
|
pem = public_pem_from_jwk(jwk_dict)
|
|
|
|
context = {
|
|
"jwk": jwk_dict,
|
|
"pem": pem,
|
|
"jwk_type": jwk_type
|
|
}
|
|
except Exception as e:
|
|
messages.error(request, _(f" {e} "))
|
|
return render(request, 'onboarding_convert_jwk.html', context)
|
|
|
|
|
|
def onboarding_convert_pem(request):
|
|
pem_type = request.GET.get('type')
|
|
context = {
|
|
"pem_type": pem_type
|
|
}
|
|
if request.method == 'POST':
|
|
try:
|
|
pem = request.POST.get('pem')
|
|
if pem_type == 'private':
|
|
jwk = private_jwk_from_pem(pem)
|
|
if pem_type == 'public':
|
|
jwk = public_jwk_from_pem(pem)
|
|
context = {
|
|
"pem": pem,
|
|
"pem_type": pem_type,
|
|
"jwk": jwk
|
|
}
|
|
except Exception as e:
|
|
messages.error(request, _(f" {e} "))
|
|
return render(request, 'onboarding_convert_pem.html', context)
|
|
|
|
|
|
def onboarding_resolve_statement(request):
|
|
if "sub" in request.GET :
|
|
form = OnboardingCreateTrustChain(request.GET)
|
|
else:
|
|
form = OnboardingCreateTrustChain()
|
|
context = {'form': form}
|
|
if form.is_valid():
|
|
context = {
|
|
'form': form,
|
|
"resolved_statement": "",
|
|
}
|
|
try:
|
|
res = resolve_entity_statement(request, format="json")
|
|
resultJson = json.loads(res.content.decode())
|
|
context["resolved_statement"] = json.dumps(resultJson, indent=4)
|
|
except Exception:
|
|
messages.error(request, _('Failed to resolve entity statement, Please check your inserted data'))
|
|
|
|
return render(request, 'onboarding_resolve_statement.html', context)
|
|
|
|
|
|
def onboarding_validating_trustmark(request):
|
|
if "id" in request.POST or "trust_mark" in request.POST:
|
|
form = OnboardingValidatingTrustMarkForm(request.POST)
|
|
else:
|
|
form = OnboardingValidatingTrustMarkForm()
|
|
|
|
context = {"form": form}
|
|
|
|
if form.is_valid():
|
|
res = trust_mark_status(request)
|
|
body = res.content.decode()
|
|
context = {'form': form}
|
|
ct = res.get("Content-Type", "")
|
|
if res.status_code == 200 and ct.startswith("application/trust-mark-status-response+jwt"):
|
|
# Draft 48: response is a signed JWT with status claim
|
|
from spid_cie_oidc.entity.jwtse import unpad_jwt_payload
|
|
content = unpad_jwt_payload(body)
|
|
active = content.get("status") == "active"
|
|
else:
|
|
# Error response (e.g. 404) or legacy JSON body
|
|
try:
|
|
content = json.loads(body)
|
|
active = content.get("active", False)
|
|
except (json.JSONDecodeError, TypeError):
|
|
active = False
|
|
if active:
|
|
messages.success(request, _('Validation Trust Mark Successfully'))
|
|
else:
|
|
messages.error(request, _('Validation Trust Mark Failed'))
|
|
return render(request, 'onboarding_validating_tm.html', context)
|
|
|
|
|
|
def onboarding_validate_md(request):
|
|
metadata_type = request.GET.get('metadata_type')
|
|
provider_profile = request.GET.get('provider_profile')
|
|
md_type = metadata_type.replace("_", " ")
|
|
title = f'Validate {md_type} {provider_profile}'
|
|
description = f'Enter a {md_type} of {provider_profile} to verify if it is compatible'
|
|
context = {
|
|
"metadata_type": metadata_type,
|
|
"provider_profile": provider_profile,
|
|
"title": title,
|
|
"description": description,
|
|
"field_name":"metadata"
|
|
}
|
|
if request.POST.get('md'):
|
|
md = request.POST['md']
|
|
context["md"] = md
|
|
try:
|
|
md_str_double_quote = md.replace("'", '"')
|
|
metadata = json.loads(md_str_double_quote)
|
|
if metadata_type == 'op_metadata':
|
|
schema = OIDCFED_PROVIDER_PROFILES[provider_profile]
|
|
schema[metadata_type](**metadata)
|
|
messages.success(request, _('Validation Metadata Successfully'))
|
|
if metadata_type == 'rp_metadata':
|
|
schema = RP_PROVIDER_PROFILES[provider_profile]
|
|
schema[metadata_type](**metadata)
|
|
messages.success(request, _('Validation Metadata Successfully'))
|
|
except Exception as e:
|
|
messages.error(request, f"Validation Failed: {e}")
|
|
return render(request, 'onboarding_validate_md.html', context)
|
|
|
|
|
|
def onboarding_validate_authn_request(request):
|
|
provider_profile = request.GET.get('provider_profile')
|
|
title = f'Validate authn request {provider_profile}'
|
|
description = f'Enter a jwt of authn request for {provider_profile} to verify if it is compatible'
|
|
context = {
|
|
"provider_profile": provider_profile,
|
|
"title": title,
|
|
"description": description,
|
|
"field_name":"jwt"
|
|
}
|
|
if request.POST.get('md'):
|
|
jwt_str = request.POST['md']
|
|
context = {
|
|
"provider_profile": provider_profile,
|
|
"title": title,
|
|
"description": description,
|
|
"field_name":"jwt",
|
|
"md": jwt_str
|
|
}
|
|
payload = unpad_jwt_payload(jwt_str)
|
|
schema = OIDCFED_PROVIDER_PROFILES[provider_profile]
|
|
try:
|
|
schema["authorization_request"](**payload)
|
|
messages.success(request, _('Validation Authn Request Successfully'))
|
|
except Exception as e:
|
|
messages.error(request, f"Validation Failed: {e}")
|
|
return render(request, 'onboarding_validate_md.html', context)
|
|
|
|
|
|
def onboarding_validate_ec(request):
|
|
context = {}
|
|
if request.POST:
|
|
url = request.POST.get("url")
|
|
context = {"url": url}
|
|
try:
|
|
ec = validate_entity_configuration(url)
|
|
context["ec"] = json.dumps(ec.payload, indent=4)
|
|
messages.success(request, _('Validation Entity Configuration Successfully'))
|
|
except Exception as e :
|
|
messages.error(request, f"Validation Failed: {e}")
|
|
return render(request, 'onboarding_validate_ec.html', context)
|
|
return render(request, 'onboarding_validate_ec.html', context)
|
|
|
|
|
|
def onboarding_decode_jwt(request):
|
|
if request.method == "GET":
|
|
form = OnboardingDecodeForm()
|
|
else:
|
|
form = OnboardingDecodeForm(request.POST)
|
|
|
|
context = {'form': form}
|
|
|
|
if form.is_valid():
|
|
form_dict = {**form.cleaned_data}
|
|
jwt = form_dict['jwt']
|
|
try:
|
|
head = unpad_jwt_head(jwt)
|
|
except Exception as e:
|
|
messages.error(
|
|
request,
|
|
f"JWS verification failed due to missing JWT header: {e}"
|
|
)
|
|
return render(request, "onboarding_decode_jwt.html", context)
|
|
|
|
if head.get('enc', None):
|
|
if not form_dict.get('jwk'):
|
|
messages.error(
|
|
request,
|
|
"JWE needs a private jwk to be decrypted"
|
|
)
|
|
return render(
|
|
request, "onboarding_decode_jwt.html", context, status = 400
|
|
)
|
|
# JWE here
|
|
jwk = form_dict['jwk']
|
|
jwt = decrypt_jwe(jwt, jwk)
|
|
|
|
try:
|
|
|
|
payload = unpad_jwt_payload(jwt)
|
|
context["head"] = json.dumps(head, indent=4)
|
|
context["payload"] = json.dumps(payload, indent=4)
|
|
jwk = form_dict["jwk"]
|
|
if jwk:
|
|
verify_jws(jwt, jwk)
|
|
messages.success(request, _('Your jws is verified'))
|
|
except Exception as e:
|
|
messages.error(request, f"JWS verification failed: {e}")
|
|
|
|
return render(request, "onboarding_decode_jwt.html", context)
|
|
|
|
|
|
def onboarding_apply_policy(request):
|
|
context = {
|
|
"md": "",
|
|
"policy": "",
|
|
"result": ""
|
|
}
|
|
if request.GET.get('md') and request.GET.get('policy'):
|
|
context = {
|
|
"md": request.GET['md'],
|
|
"policy": request.GET['policy'],
|
|
"result": ""
|
|
}
|
|
try:
|
|
md = json.loads(request.GET['md'])
|
|
policy = json.loads(request.GET['policy'])
|
|
reuslt = apply_policy(md, policy)
|
|
context["result"] = json.dumps(reuslt, indent=4)
|
|
except Exception as e:
|
|
messages.error(request, {e})
|
|
return render(request, 'onboarding_apply_policy.html', context)
|
|
|
|
|
|
def onboarding_schemas_authorization(request):
|
|
auth_request = AuthenticationRequestSpid.schema_json(indent=2)
|
|
auth_res_succ = AuthenticationResponse.schema_json(indent=2)
|
|
auth_res_err = AuthenticationErrorResponse.schema_json(indent=2)
|
|
content = {
|
|
"title": "Schemas authorization endpoint",
|
|
"schemas": {
|
|
"authorization request": auth_request,
|
|
"authorization successful response": auth_res_succ,
|
|
"authorization error response": auth_res_err,
|
|
}
|
|
}
|
|
return render(request, "onboarding_schemas.html", content)
|
|
|
|
|
|
def onboarding_schemas_introspection(request):
|
|
intro_request = IntrospectionRequest.schema_json(indent=2)
|
|
intro_res_succ = IntrospectionResponse.schema_json(indent=2)
|
|
intro_res_err = IntrospectionErrorResponseSpid.schema_json(indent=2)
|
|
content = {
|
|
"title": "Schemas introspection endpoint",
|
|
"schemas": {
|
|
"introspection request": intro_request,
|
|
"introspection successful response": intro_res_succ,
|
|
"introspection error response": intro_res_err,
|
|
}
|
|
}
|
|
return render(request, "onboarding_schemas.html", content)
|
|
|
|
|
|
def onboarding_schemas_federation_entity_endpoints(request):
|
|
federation_entity_fetch_request = FetchRequest.schema_json(indent=2)
|
|
federation_entity_list_request = ListRequest.schema_json(indent=2)
|
|
federation_entity_list_response = ListResponse.schema_json(indent=2)
|
|
federation_entity_resolve_request = ResolveRequest.schema_json(indent=2)
|
|
federation_entity_resolve_response = ResolveResponse.schema_json(indent=2)
|
|
federation_entity_status_request = TrustMarkRequest.schema_json(indent=2)
|
|
federation_entity_status_response = TrustMarkResponse.schema_json(indent=2)
|
|
content = {
|
|
"title": "Schemas Federation entity endpoint",
|
|
"schemas": {
|
|
"Federation entity fetch": federation_entity_fetch_request,
|
|
"Federation entity list request": federation_entity_list_request,
|
|
"Federation entity list response": federation_entity_list_response,
|
|
"Federation entity resolve request": federation_entity_resolve_request,
|
|
"Federation entity resolve response": federation_entity_resolve_response,
|
|
"Federation entity status request": federation_entity_status_request,
|
|
"Federation entity status response": federation_entity_status_response
|
|
}
|
|
}
|
|
return render(request, "onboarding_schemas.html", content)
|
|
|
|
|
|
def onboarding_schemas_metadata(request):
|
|
op_meta_spid = OPMetadataSpid.schema_json(indent=2)
|
|
op_meta_cie = OPMetadataCie.schema_json(indent=2)
|
|
rp_meta_spid = RPMetadataSpid.schema_json(indent=2)
|
|
rp_meta_cie = RPMetadataCie.schema_json(indent=2)
|
|
fed_entity = FAMetadata.schema_json(indent=2)
|
|
content = {
|
|
"title": "Schemas metadata Spid/Cie",
|
|
"schemas": {
|
|
"Federation entity": fed_entity,
|
|
"OP Spid": op_meta_spid,
|
|
"OP Cie": op_meta_cie,
|
|
"RP Spid": rp_meta_spid,
|
|
"RP Cie": rp_meta_cie
|
|
}
|
|
}
|
|
return render(request, "onboarding_schemas.html", content)
|
|
|
|
|
|
def onboarding_schemas_revocation(request):
|
|
revoc_request = RevocationRequest.schema_json(indent=2)
|
|
content = {
|
|
"title": "Schemas revocation endpoint",
|
|
"schemas": {
|
|
"revocation request": revoc_request
|
|
}
|
|
}
|
|
return render(request, "onboarding_schemas.html", content)
|
|
|
|
|
|
def onboarding_schemas_token(request):
|
|
auth_code_req = TokenAuthnCodeRequest.schema_json(indent=2)
|
|
auth_code_res = TokenResponse.schema_json(indent=2)
|
|
refr_req = TokenRefreshRequest.schema_json(indent=2)
|
|
refr_res = TokenRefreshResponse.schema_json(indent=2)
|
|
err_res = TokenErrorResponse.schema_json(indent=2)
|
|
content = {
|
|
"title": "Schemas token endpoint",
|
|
"schemas": {
|
|
"token authn code request": auth_code_req,
|
|
"token authn code response": auth_code_res,
|
|
"refresh token request": refr_req,
|
|
"refresh token response": refr_res,
|
|
"token error response": err_res
|
|
}
|
|
}
|
|
return render(request, "onboarding_schemas.html", content)
|
|
|
|
|
|
def onboarding_schemas_jwt_client_assertion(request):
|
|
jwt_client_asser = JwtStructure.schema_json(indent=2)
|
|
content = {
|
|
"title": "Schemas jwt",
|
|
"schemas": {
|
|
"jwt client assertion": jwt_client_asser
|
|
}
|
|
}
|
|
return render(request, "onboarding_schemas.html", content)
|