253 lines
8.2 KiB
Python
253 lines
8.2 KiB
Python
from django.conf import settings
|
|
from django.utils import timezone
|
|
from rest_framework_simplejwt.tokens import AccessToken
|
|
import os
|
|
import secrets
|
|
from datetime import timedelta
|
|
|
|
from django.db.models import F
|
|
|
|
from apps.authentication.models import RecoveryCode
|
|
from apps.common.utils import client_ip, device_name, generate_token, hash_token
|
|
from apps.identity.models import User
|
|
from apps.oauth.models import AccessToken as OAuthAccessToken, RefreshToken
|
|
from apps.session.models import Session, SessionStatus, SessionType
|
|
from apps.token_blacklist.models import TokenBlacklist
|
|
|
|
|
|
def _get_jwt_algorithm():
|
|
"""Return the JWT algorithm from settings (default HS256)."""
|
|
return settings.SIMPLE_JWT.get("ALGORITHM", "HS256")
|
|
|
|
|
|
def _get_jwt_signing_key():
|
|
"""Return the appropriate signing key based on algorithm."""
|
|
algorithm = _get_jwt_algorithm()
|
|
if algorithm in ("RS256", "RS384", "RS512"):
|
|
key_path = settings.SIMPLE_JWT.get("JWT_PRIVATE_KEY_PATH")
|
|
if key_path and os.path.exists(key_path):
|
|
with open(key_path, "rb") as f:
|
|
return f.read()
|
|
return settings.SIMPLE_JWT["SIGNING_KEY"]
|
|
|
|
|
|
def _get_jwt_verifying_key():
|
|
"""Return the appropriate verifying key based on algorithm."""
|
|
algorithm = _get_jwt_algorithm()
|
|
if algorithm in ("RS256", "RS384", "RS512"):
|
|
key_path = settings.SIMPLE_JWT.get("JWT_PUBLIC_KEY_PATH")
|
|
if key_path and os.path.exists(key_path):
|
|
with open(key_path, "rb") as f:
|
|
return f.read()
|
|
return settings.SIMPLE_JWT["SIGNING_KEY"]
|
|
|
|
|
|
def create_access_token(user, product_key=None, organization_id=None, scopes=None):
|
|
from apps.verification.services import TrustEngine
|
|
from apps.verification.models import TrustState
|
|
|
|
algorithm = _get_jwt_algorithm()
|
|
signing_key = _get_jwt_signing_key()
|
|
verifying_key = _get_jwt_verifying_key()
|
|
|
|
trust_state = TrustEngine.compute_trust_state(user)
|
|
trust_score = TrustEngine.get_trust_score(user)
|
|
amr = list(
|
|
TrustEngine.get_person_evidence(user).values_list("evidence_type", flat=True)
|
|
) or ["pwd"]
|
|
|
|
access = AccessToken()
|
|
access.payload.update(
|
|
{
|
|
"user_id": str(user.id),
|
|
"user_status": user.status,
|
|
"identity_key": user.identity_key,
|
|
"token_type": "access",
|
|
"trust_state": trust_state,
|
|
"trust_score": trust_score,
|
|
"acr": f"urn:openid:params:acr:{trust_state}",
|
|
"amr": amr,
|
|
"token_version": user.token_version,
|
|
"scope": " ".join(scopes) if scopes else "openid",
|
|
}
|
|
)
|
|
if product_key:
|
|
access.payload["product_key"] = product_key
|
|
if organization_id:
|
|
access.payload["organization_id"] = str(organization_id)
|
|
return access
|
|
|
|
|
|
def create_auth_tokens(
|
|
user, request, application=None, product_key=None, organization_id=None
|
|
):
|
|
refresh_lifetime = settings.SIMPLE_JWT["REFRESH_TOKEN_LIFETIME"]
|
|
user_agent = request.META.get("HTTP_USER_AGENT", "")[:500]
|
|
|
|
session = Session.objects.create(
|
|
user=user,
|
|
application=application,
|
|
session_type=SessionType.BROWSER,
|
|
ip_address=client_ip(request),
|
|
user_agent=user_agent,
|
|
device_name=device_name(user_agent),
|
|
expires_at=timezone.now() + refresh_lifetime,
|
|
)
|
|
|
|
raw_refresh = generate_token()
|
|
RefreshToken.objects.create(
|
|
user=user,
|
|
application=application,
|
|
session=session,
|
|
token_hash=hash_token(raw_refresh),
|
|
expires_at=session.expires_at,
|
|
)
|
|
|
|
access = create_access_token(
|
|
user, product_key=product_key, organization_id=organization_id
|
|
)
|
|
|
|
return {
|
|
"access": str(access),
|
|
"refresh": raw_refresh,
|
|
"expires_in": int(settings.SIMPLE_JWT["ACCESS_TOKEN_LIFETIME"].total_seconds()),
|
|
"token_type": "bearer",
|
|
"session_id": str(session.id),
|
|
}
|
|
|
|
|
|
def rotate_refresh_tokens(user, session, application=None):
|
|
RefreshToken.objects.filter(session=session, revoked=False).update(
|
|
revoked=True,
|
|
replaced_by="rotated",
|
|
)
|
|
raw_refresh = generate_token()
|
|
new_token = RefreshToken.objects.create(
|
|
user=user,
|
|
application=application,
|
|
session=session,
|
|
token_hash=hash_token(raw_refresh),
|
|
expires_at=timezone.now() + settings.SIMPLE_JWT["REFRESH_TOKEN_LIFETIME"],
|
|
)
|
|
return raw_refresh, new_token
|
|
|
|
|
|
def blacklist_token(token, token_type="access"):
|
|
"""Add token to blacklist for instant revocation checking."""
|
|
token_hash = hash_token(token)
|
|
access_lifetime = settings.SIMPLE_JWT["ACCESS_TOKEN_LIFETIME"]
|
|
expires_at = timezone.now() + access_lifetime
|
|
TokenBlacklist.objects.update_or_create(
|
|
token_hash=token_hash,
|
|
defaults={
|
|
"token_type": token_type,
|
|
"expires_at": expires_at,
|
|
},
|
|
)
|
|
|
|
|
|
def is_token_blacklisted(token):
|
|
"""Check if token is in blacklist."""
|
|
token_hash = hash_token(token)
|
|
try:
|
|
entry = TokenBlacklist.objects.get(token_hash=token_hash)
|
|
return not entry.is_expired
|
|
except TokenBlacklist.DoesNotExist:
|
|
return False
|
|
|
|
|
|
def _flush_django_sessions(user):
|
|
"""Delete every Django browser session (the ``sessionid`` SSO cookie) for
|
|
the user so a suspended or globally-logged-out account can no longer use
|
|
single sign-on on any device."""
|
|
from django.contrib.sessions.models import Session as DjangoSession
|
|
|
|
target = str(user.id)
|
|
for session in DjangoSession.objects.all():
|
|
try:
|
|
if session.get_decoded().get("_auth_user_id") == target:
|
|
session.delete()
|
|
except Exception:
|
|
continue
|
|
|
|
|
|
def revoke_all_user_sessions(user, reason="global_logout"):
|
|
"""Global (SSO) termination: revoke every active session, refresh token
|
|
and OAuth access token for the user, then bump ``token_version`` so any
|
|
outstanding stateless JWT access tokens are rejected on next use.
|
|
|
|
Used by global logout (decision 8) and by suspend enforcement (decision 9).
|
|
"""
|
|
now = timezone.now()
|
|
Session.objects.filter(user=user, status=SessionStatus.ACTIVE).update(
|
|
status=SessionStatus.REVOKED,
|
|
revoked_at=now,
|
|
revoked_reason=reason,
|
|
)
|
|
RefreshToken.objects.filter(user=user, revoked=False).update(
|
|
revoked=True,
|
|
replaced_by="terminated",
|
|
)
|
|
OAuthAccessToken.objects.filter(user=user, revoked=False).update(revoked=True)
|
|
_flush_django_sessions(user)
|
|
User.objects.filter(pk=user.pk).update(token_version=F("token_version") + 1)
|
|
|
|
|
|
def generate_recovery_codes(user, count=10, ttl_days=365):
|
|
"""Generate a fresh set of one-time MFA recovery codes (Google-style).
|
|
|
|
Any previously unused codes are invalidated. Returns the plaintext codes
|
|
(shown to the user once); only their hashes are persisted.
|
|
"""
|
|
RECOVERY_PURPOSE = "mfa_bypass"
|
|
RecoveryCode.objects.filter(
|
|
user=user, purpose=RECOVERY_PURPOSE, used=False
|
|
).delete()
|
|
|
|
now = timezone.now()
|
|
expires_at = now + timedelta(days=ttl_days)
|
|
plaintext = []
|
|
for _ in range(count):
|
|
raw = secrets.token_hex(4).upper()
|
|
code = f"{raw[:4]}-{raw[4:]}"
|
|
RecoveryCode.objects.create(
|
|
user=user,
|
|
purpose=RECOVERY_PURPOSE,
|
|
code_hash=hash_token(code),
|
|
expires_at=expires_at,
|
|
)
|
|
plaintext.append(code)
|
|
return plaintext
|
|
|
|
|
|
def count_remaining_recovery_codes(user):
|
|
RECOVERY_PURPOSE = "mfa_bypass"
|
|
return RecoveryCode.objects.filter(
|
|
user=user,
|
|
purpose=RECOVERY_PURPOSE,
|
|
used=False,
|
|
expires_at__gt=timezone.now(),
|
|
).count()
|
|
|
|
|
|
def redeem_recovery_code(user, code):
|
|
"""Consume a single recovery code. Returns True if it was valid + unused."""
|
|
RECOVERY_PURPOSE = "mfa_bypass"
|
|
if not code:
|
|
return False
|
|
normalized = code.strip().upper()
|
|
code_hash = hash_token(normalized)
|
|
rc = RecoveryCode.objects.filter(
|
|
user=user,
|
|
purpose=RECOVERY_PURPOSE,
|
|
code_hash=code_hash,
|
|
used=False,
|
|
expires_at__gt=timezone.now(),
|
|
).first()
|
|
if not rc:
|
|
return False
|
|
rc.used = True
|
|
rc.used_at = timezone.now()
|
|
rc.save(update_fields=["used", "used_at", "updated_at"])
|
|
return True
|