105 lines
4.2 KiB
Python
105 lines
4.2 KiB
Python
|
"""Same-origin identity contract in the existing Django service, deny by default."""
|
||
|
|
from functools import wraps
|
||
|
|
|
||
|
|
from django.contrib.auth import get_user_model
|
||
|
|
from django.db import DatabaseError
|
||
|
|
from django.http import JsonResponse
|
||
|
|
from django.views.decorators.http import require_GET, require_POST, require_http_methods
|
||
|
|
|
||
|
|
from django_ui.views import AUDIT, client_ip, json_body, security_headers
|
||
|
|
from .models import APPLICATIONS, ROLES, ApplicationGrant
|
||
|
|
|
||
|
|
PERMISSIONS = {"reader": ("read",), "operator": ("read", "write"), "admin": ("read", "write", "admin")}
|
||
|
|
|
||
|
|
|
||
|
|
def reply(body, status=200):
|
||
|
|
response = security_headers(JsonResponse(body, status=status))
|
||
|
|
response["Vary"] = "Cookie"
|
||
|
|
return response
|
||
|
|
|
||
|
|
|
||
|
|
def identity_request(view):
|
||
|
|
@wraps(view)
|
||
|
|
def guarded(request, application, **kwargs):
|
||
|
|
try:
|
||
|
|
if not request.user.is_authenticated:
|
||
|
|
return reply({"detail": "Authentification requise"}, 401)
|
||
|
|
if application not in APPLICATIONS:
|
||
|
|
return reply({"detail": "Application inconnue"}, 404)
|
||
|
|
if request.headers.get("Sec-Fetch-Site") == "cross-site":
|
||
|
|
return reply({"detail": "Requête intersite refusée"}, 403)
|
||
|
|
if request.method not in {"GET", "HEAD", "OPTIONS"} and (
|
||
|
|
request.content_type != "application/json" or len(request.body) > 4096
|
||
|
|
):
|
||
|
|
return reply({"detail": "Corps JSON limité requis"}, 400)
|
||
|
|
return view(request, application, **kwargs)
|
||
|
|
except DatabaseError:
|
||
|
|
return reply({"detail": "Identité indisponible ; accès refusé"}, 503)
|
||
|
|
|
||
|
|
return guarded
|
||
|
|
|
||
|
|
|
||
|
|
def principal(request, application):
|
||
|
|
grant = ApplicationGrant.objects.filter(user=request.user, application=application).first()
|
||
|
|
if not grant or grant.role not in PERMISSIONS:
|
||
|
|
return None
|
||
|
|
return {
|
||
|
|
"issuer": "django-codexpi",
|
||
|
|
"subject": str(request.user.pk),
|
||
|
|
"username": request.user.get_username(),
|
||
|
|
"application": application,
|
||
|
|
"role": grant.role,
|
||
|
|
"permissions": list(PERMISSIONS[grant.role]),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@require_GET
|
||
|
|
@identity_request
|
||
|
|
def application_identity(request, application):
|
||
|
|
identity = principal(request, application)
|
||
|
|
return reply(identity) if identity else reply({"detail": "Accès à cette application refusé"}, 403)
|
||
|
|
|
||
|
|
|
||
|
|
@require_POST
|
||
|
|
@identity_request
|
||
|
|
def authorize(request, application):
|
||
|
|
body = json_body(request)
|
||
|
|
if set(body) != {"permission"} or body["permission"] not in ("read", "write", "admin"):
|
||
|
|
return reply({"detail": "Permission invalide"}, 400)
|
||
|
|
identity = principal(request, application)
|
||
|
|
if not identity or body["permission"] not in identity["permissions"]:
|
||
|
|
return reply({"detail": "Permission refusée"}, 403)
|
||
|
|
return reply({**identity, "authorized_permission": body["permission"]})
|
||
|
|
|
||
|
|
|
||
|
|
@require_http_methods(["GET", "PUT", "DELETE"])
|
||
|
|
@identity_request
|
||
|
|
def application_grant(request, application, user_id):
|
||
|
|
if not (request.user.is_staff or request.user.is_superuser):
|
||
|
|
return reply({"detail": "Administration des identités requise"}, 403)
|
||
|
|
target = get_user_model().objects.filter(pk=user_id).first()
|
||
|
|
if not target:
|
||
|
|
return reply({"detail": "Compte introuvable"}, 404)
|
||
|
|
grants = ApplicationGrant.objects.filter(user=target, application=application)
|
||
|
|
if request.method == "GET":
|
||
|
|
grant = grants.first()
|
||
|
|
return reply({"subject": str(target.pk), "application": application, "role": grant.role if grant else None})
|
||
|
|
if request.method == "DELETE":
|
||
|
|
grants.delete()
|
||
|
|
role = None
|
||
|
|
else:
|
||
|
|
body = json_body(request)
|
||
|
|
if set(body) != {"role"} or body["role"] not in ROLES:
|
||
|
|
return reply({"detail": "Rôle invalide"}, 400)
|
||
|
|
role = body["role"]
|
||
|
|
ApplicationGrant.objects.update_or_create(user=target, application=application, defaults={"role": role})
|
||
|
|
AUDIT.log(
|
||
|
|
"identity.grant_revoked" if role is None else "identity.grant_updated",
|
||
|
|
client_ip(request),
|
||
|
|
username=request.user.get_username(),
|
||
|
|
project_id=application,
|
||
|
|
outcome="allowed",
|
||
|
|
reason=f"target_user={target.pk};role={role or 'none'}",
|
||
|
|
)
|
||
|
|
return reply({"subject": str(target.pk), "application": application, "role": role})
|