648 lines
22 KiB
Python
648 lines
22 KiB
Python
import base64
|
|
import hashlib
|
|
import logging
|
|
import os
|
|
import secrets
|
|
from urllib.parse import urlencode
|
|
from urllib.parse import urlparse
|
|
|
|
import jwt
|
|
import requests
|
|
from django.core.cache import cache
|
|
from django.http import HttpResponseRedirect
|
|
from django.utils.crypto import get_random_string
|
|
from django.utils import timezone
|
|
from django.views import View
|
|
|
|
from plane.authentication.utils.host import base_host
|
|
from plane.authentication.utils.login import user_login
|
|
from plane.authentication.utils.redirection_path import get_redirection_path
|
|
from plane.authentication.nodedc_workspace_join import (
|
|
NodeDCWorkspaceInviteJoinError,
|
|
accept_nodedc_workspace_invite_request_for_user,
|
|
)
|
|
from plane.authentication.views.nodedc_logout import get_nodedc_internal_token
|
|
from plane.db.models import ExternalIdentityLink, Profile, User
|
|
from plane.utils.path_validator import get_safe_redirect_url, validate_next_path
|
|
|
|
|
|
OIDC_SESSION_KEY = "nodedc_oidc"
|
|
OIDC_PROVIDER = "authentik"
|
|
NODEDC_WORKSPACE_INVITE_ACCEPT_PREFIX = "/auth/nodedc/workspace-invite/accept/"
|
|
DEFAULT_REQUIRED_GROUPS = "nodedc:superadmin,nodedc:taskmanager:admin,nodedc:taskmanager:user"
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class NodeDCOIDCInitiateEndpoint(View):
|
|
def get(self, request):
|
|
config = get_oidc_config()
|
|
next_path = validate_next_path(request.GET.get("next_path", ""))
|
|
discovery = load_discovery(config["issuer"])
|
|
state = secrets.token_urlsafe(32)
|
|
nonce = secrets.token_urlsafe(32)
|
|
code_verifier = secrets.token_urlsafe(64)
|
|
code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest()).decode().rstrip("=")
|
|
|
|
request.session[OIDC_SESSION_KEY] = {
|
|
"state": state,
|
|
"nonce": nonce,
|
|
"code_verifier": code_verifier,
|
|
"next_path": next_path,
|
|
}
|
|
request.session.save()
|
|
|
|
params = {
|
|
"response_type": "code",
|
|
"client_id": config["client_id"],
|
|
"redirect_uri": config["redirect_uri"],
|
|
"scope": config["scope"],
|
|
"state": state,
|
|
"nonce": nonce,
|
|
"code_challenge": code_challenge,
|
|
"code_challenge_method": "S256",
|
|
}
|
|
|
|
if request.GET.get("prompt") == "login":
|
|
params["prompt"] = "login"
|
|
params["max_age"] = "0"
|
|
|
|
return HttpResponseRedirect(f"{discovery['authorization_endpoint']}?{urlencode(params)}")
|
|
|
|
|
|
class NodeDCOIDCCallbackEndpoint(View):
|
|
def get(self, request):
|
|
config = get_oidc_config()
|
|
oidc_session = request.session.get(OIDC_SESSION_KEY) or {}
|
|
next_path = oidc_session.get("next_path", "")
|
|
base_url = base_host(request=request, is_app=True)
|
|
|
|
if request.GET.get("error"):
|
|
request.session.pop(OIDC_SESSION_KEY, None)
|
|
return oidc_login_redirect(base_url, next_path)
|
|
|
|
state = request.GET.get("state")
|
|
code = request.GET.get("code")
|
|
|
|
if not state or state != oidc_session.get("state") or not code:
|
|
request.session.pop(OIDC_SESSION_KEY, None)
|
|
return oidc_login_redirect(base_url, next_path)
|
|
|
|
try:
|
|
discovery = load_discovery(config["issuer"])
|
|
token_set = exchange_code(discovery, config, code, oidc_session.get("code_verifier"))
|
|
claims = verify_id_token(discovery, config, token_set["id_token"], oidc_session.get("nonce"))
|
|
except (KeyError, RuntimeError, requests.RequestException, jwt.PyJWTError):
|
|
request.session.pop(OIDC_SESSION_KEY, None)
|
|
return oidc_login_redirect(base_url, next_path)
|
|
|
|
if is_logout_guard_active(claims):
|
|
request.session.pop(OIDC_SESSION_KEY, None)
|
|
return oidc_login_redirect(base_url, next_path, prompt_login=True)
|
|
|
|
groups = normalize_groups(claims.get("groups"))
|
|
|
|
if not has_required_group(groups):
|
|
return oidc_error_redirect(base_url, next_path, "oidc_access_denied")
|
|
|
|
user = resolve_linked_user(
|
|
claims=claims,
|
|
groups=groups,
|
|
auto_link=config["auto_link_email"],
|
|
auto_create=config["auto_create_user"],
|
|
sync_profile=config["sync_profile"],
|
|
skip_profile_onboarding=config["skip_profile_onboarding"],
|
|
)
|
|
|
|
if user is None or not user.is_active:
|
|
return oidc_error_redirect(base_url, next_path, "oidc_user_not_linked")
|
|
|
|
request.session.pop(OIDC_SESSION_KEY, None)
|
|
user_login(request=request, user=user, is_app=True)
|
|
|
|
path = next_path or get_redirection_path(user=user)
|
|
return HttpResponseRedirect(get_safe_redirect_url(base_url=base_url, next_path=path, params={}))
|
|
|
|
|
|
class NodeDCHandoffEndpoint(View):
|
|
def get(self, request):
|
|
config = get_oidc_config()
|
|
base_url = base_host(request=request, is_app=True)
|
|
next_path = validate_next_path(request.GET.get("next_path", ""))
|
|
token = request.GET.get("token", "")
|
|
|
|
try:
|
|
handoff = consume_launcher_handoff(token)
|
|
except (RuntimeError, requests.RequestException, ValueError):
|
|
logger.warning("NODEDC handoff failed", exc_info=True)
|
|
return oidc_login_redirect(base_url, next_path)
|
|
|
|
handoff_user = handoff.get("user") or {}
|
|
groups = normalize_groups(handoff_user.get("groups"))
|
|
|
|
if not has_required_group(groups):
|
|
return oidc_error_redirect(base_url, next_path, "handoff_access_denied")
|
|
|
|
claims = {
|
|
"sub": str(handoff_user.get("subject") or handoff_user.get("authentikUserId") or handoff_user.get("id") or ""),
|
|
"email": str(handoff_user.get("email") or "").strip().lower(),
|
|
"name": handoff_user.get("name"),
|
|
"preferred_username": handoff_user.get("email"),
|
|
"picture": handoff_user.get("avatarUrl"),
|
|
"groups": groups,
|
|
"email_verified": True,
|
|
}
|
|
|
|
user = resolve_linked_user(
|
|
claims=claims,
|
|
groups=groups,
|
|
auto_link=config["auto_link_email"],
|
|
auto_create=config["auto_create_user"],
|
|
sync_profile=config["sync_profile"],
|
|
skip_profile_onboarding=config["skip_profile_onboarding"],
|
|
)
|
|
|
|
if user is None or not user.is_active:
|
|
return oidc_error_redirect(base_url, next_path, "handoff_user_not_linked")
|
|
|
|
user_login(request=request, user=user, is_app=True)
|
|
|
|
workspace_invite_redirect = resolve_nodedc_workspace_invite_accept_redirect(base_url, next_path, user)
|
|
if workspace_invite_redirect is not None:
|
|
return workspace_invite_redirect
|
|
|
|
path = next_path or get_redirection_path(user=user)
|
|
return HttpResponseRedirect(get_safe_redirect_url(base_url=base_url, next_path=path, params={}))
|
|
|
|
|
|
def resolve_nodedc_workspace_invite_accept_redirect(base_url, next_path, user):
|
|
if not next_path.startswith(NODEDC_WORKSPACE_INVITE_ACCEPT_PREFIX):
|
|
return None
|
|
|
|
request_id = next_path.removeprefix(NODEDC_WORKSPACE_INVITE_ACCEPT_PREFIX).strip("/")
|
|
|
|
try:
|
|
result = accept_nodedc_workspace_invite_request_for_user(request_id, user)
|
|
except NodeDCWorkspaceInviteJoinError as error:
|
|
logger.warning(
|
|
"NODEDC managed workspace invite accept failed: code=%s user_id=%s request_id=%s",
|
|
error.code,
|
|
user.id,
|
|
request_id,
|
|
)
|
|
return oidc_error_redirect(base_url, "", error.code)
|
|
except Exception:
|
|
logger.exception(
|
|
"NODEDC managed workspace invite accept crashed: user_id=%s request_id=%s",
|
|
user.id,
|
|
request_id,
|
|
)
|
|
return oidc_error_redirect(base_url, "", "workspace_invite_accept_failed")
|
|
|
|
workspace_slug = result.get("workspaceSlug")
|
|
if not workspace_slug:
|
|
return oidc_error_redirect(base_url, "", "workspace_invite_redirect_missing")
|
|
|
|
return HttpResponseRedirect(f"{base_url.rstrip('/')}/{workspace_slug}/")
|
|
|
|
|
|
def get_oidc_config():
|
|
issuer = os.environ.get("PLANE_OIDC_ISSUER", "").strip()
|
|
client_id = os.environ.get("PLANE_OIDC_CLIENT_ID", "").strip()
|
|
client_secret = os.environ.get("PLANE_OIDC_CLIENT_SECRET", "").strip()
|
|
redirect_uri = os.environ.get("PLANE_OIDC_REDIRECT_URI", "").strip()
|
|
|
|
if not issuer or not client_id or not client_secret or not redirect_uri:
|
|
raise RuntimeError("Plane OIDC is not configured")
|
|
|
|
return {
|
|
"issuer": issuer.rstrip("/") + "/",
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
"redirect_uri": redirect_uri,
|
|
"scope": os.environ.get("PLANE_OIDC_SCOPE", "openid email profile groups"),
|
|
"auto_link_email": os.environ.get("PLANE_OIDC_AUTO_LINK_EMAIL", "0") == "1",
|
|
"auto_create_user": os.environ.get("PLANE_OIDC_AUTO_CREATE_USER", "0") == "1",
|
|
"sync_profile": os.environ.get("PLANE_OIDC_SYNC_PROFILE", "1") == "1",
|
|
"skip_profile_onboarding": os.environ.get("PLANE_NODEDC_SKIP_PROFILE_ONBOARDING", "0") == "1",
|
|
"handoff_url": os.environ.get(
|
|
"PLANE_NODEDC_HANDOFF_URL",
|
|
"http://launcher.local.nodedc/api/internal/handoff/consume",
|
|
).strip(),
|
|
}
|
|
|
|
|
|
def consume_launcher_handoff(token):
|
|
if not token:
|
|
raise RuntimeError("NODEDC handoff token is missing")
|
|
|
|
config = get_oidc_config()
|
|
internal_token = get_nodedc_internal_token()
|
|
|
|
if not config["handoff_url"] or not internal_token:
|
|
raise RuntimeError("NODEDC handoff is not configured")
|
|
|
|
response = requests.post(
|
|
config["handoff_url"],
|
|
json={
|
|
"token": token,
|
|
"serviceSlug": os.environ.get("PLANE_NODEDC_ACCESS_SERVICE_SLUG", "task-manager"),
|
|
},
|
|
headers={"Authorization": f"Bearer {internal_token}"},
|
|
timeout=float(os.environ.get("PLANE_NODEDC_HANDOFF_TIMEOUT_SECONDS", "3")),
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
|
|
if not payload.get("ok"):
|
|
raise RuntimeError("NODEDC handoff was rejected")
|
|
|
|
return payload
|
|
|
|
|
|
def load_discovery(issuer):
|
|
response = requests.get(f"{issuer}.well-known/openid-configuration", timeout=10)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def exchange_code(discovery, config, code, code_verifier):
|
|
response = requests.post(
|
|
discovery["token_endpoint"],
|
|
data={
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": config["redirect_uri"],
|
|
"code_verifier": code_verifier,
|
|
},
|
|
auth=(config["client_id"], config["client_secret"]),
|
|
timeout=10,
|
|
)
|
|
response.raise_for_status()
|
|
token_set = response.json()
|
|
|
|
if not token_set.get("id_token"):
|
|
raise RuntimeError("OIDC token response does not contain id_token")
|
|
|
|
return token_set
|
|
|
|
|
|
def verify_id_token(discovery, config, id_token, nonce):
|
|
jwks_client = jwt.PyJWKClient(discovery["jwks_uri"])
|
|
signing_key = jwks_client.get_signing_key_from_jwt(id_token)
|
|
claims = jwt.decode(
|
|
id_token,
|
|
signing_key.key,
|
|
algorithms=["RS256"],
|
|
audience=config["client_id"],
|
|
issuer=discovery.get("issuer", config["issuer"]),
|
|
)
|
|
|
|
if claims.get("nonce") != nonce:
|
|
raise RuntimeError("OIDC nonce validation failed")
|
|
|
|
return claims
|
|
|
|
|
|
def normalize_groups(groups):
|
|
if isinstance(groups, list):
|
|
return list(dict.fromkeys(group for group in groups if isinstance(group, str)))
|
|
if isinstance(groups, str) and groups:
|
|
return [groups]
|
|
return []
|
|
|
|
|
|
def has_required_group(groups):
|
|
required_groups = {
|
|
group.strip()
|
|
for group in os.environ.get("PLANE_OIDC_REQUIRED_GROUPS", DEFAULT_REQUIRED_GROUPS).split(",")
|
|
if group.strip()
|
|
}
|
|
return bool(required_groups.intersection(set(groups)))
|
|
|
|
|
|
def resolve_linked_user(claims, groups, auto_link, auto_create, sync_profile, skip_profile_onboarding):
|
|
subject = str(claims.get("sub") or "")
|
|
email = str(claims.get("email") or "").strip().lower()
|
|
|
|
if not subject or not email:
|
|
return None
|
|
|
|
link = ExternalIdentityLink.objects.select_related("user").filter(
|
|
provider=OIDC_PROVIDER,
|
|
subject=subject,
|
|
).first()
|
|
|
|
if link and link.status != ExternalIdentityLink.Status.ACTIVE:
|
|
logger.warning(
|
|
"NODEDC OIDC denied disabled external identity link: provider=%s subject_hash=%s",
|
|
OIDC_PROVIDER,
|
|
hash_subject(subject),
|
|
)
|
|
return None
|
|
|
|
if link is None and auto_link and email:
|
|
user = User.objects.filter(email__iexact=email, is_active=True).first()
|
|
if user:
|
|
link, _ = ExternalIdentityLink.objects.get_or_create(
|
|
provider=OIDC_PROVIDER,
|
|
subject=subject,
|
|
defaults={"user": user, "email": email, "groups": groups},
|
|
)
|
|
|
|
if link is None and auto_create and email:
|
|
user, user_created = get_or_create_oidc_user(email=email, claims=claims)
|
|
link, _ = ExternalIdentityLink.objects.get_or_create(
|
|
provider=OIDC_PROVIDER,
|
|
subject=subject,
|
|
defaults={"user": user, "email": email, "groups": groups},
|
|
)
|
|
if user_created:
|
|
logger.info(
|
|
"NODEDC OIDC provisioned Tasker user: user_id=%s email_hash=%s subject_hash=%s",
|
|
user.id,
|
|
hash_email(email),
|
|
hash_subject(subject),
|
|
)
|
|
|
|
if link is None:
|
|
logger.warning(
|
|
"NODEDC OIDC user is not linked: provider=%s email_hash=%s subject_hash=%s",
|
|
OIDC_PROVIDER,
|
|
hash_email(email),
|
|
hash_subject(subject),
|
|
)
|
|
return None
|
|
|
|
link.email = email or link.email
|
|
link.groups = groups
|
|
link.last_login_at = timezone.now()
|
|
link.save(update_fields=["email", "groups", "last_login_at", "updated_at"])
|
|
|
|
user = link.user
|
|
user.last_login_medium = OIDC_PROVIDER
|
|
user.last_login_time = timezone.now()
|
|
update_fields = ["last_login_medium", "last_login_time", "updated_at"]
|
|
|
|
if sync_profile:
|
|
update_fields.extend(sync_user_profile_from_claims(user, claims))
|
|
|
|
user.save(update_fields=list(dict.fromkeys(update_fields)))
|
|
|
|
if skip_profile_onboarding:
|
|
ensure_nodedc_profile_onboarded(user)
|
|
|
|
return user
|
|
|
|
|
|
def get_or_create_oidc_user(email, claims):
|
|
user = User.objects.filter(email__iexact=email).first()
|
|
|
|
if user:
|
|
return user, False
|
|
|
|
username = build_unique_username(email)
|
|
display_name = first_string_claim(claims, "name", "preferred_username") or User.get_display_name(email)
|
|
given_name = first_string_claim(claims, "given_name") or ""
|
|
family_name = first_string_claim(claims, "family_name") or ""
|
|
avatar_url = normalize_nodedc_avatar_url(first_string_claim(claims, "picture", "avatar_url", "avatar") or "")
|
|
|
|
user = User.objects.create_user(
|
|
email=email,
|
|
username=username,
|
|
display_name=display_name,
|
|
first_name=given_name,
|
|
last_name=family_name,
|
|
avatar=avatar_url,
|
|
is_active=True,
|
|
is_managed=True,
|
|
is_password_autoset=True,
|
|
is_email_verified=bool(claims.get("email_verified", False)),
|
|
last_login_medium=OIDC_PROVIDER,
|
|
last_login_time=timezone.now(),
|
|
)
|
|
return user, True
|
|
|
|
|
|
def ensure_nodedc_profile_onboarded(user):
|
|
profile, _ = Profile.objects.get_or_create(user=user)
|
|
onboarding_step = dict(profile.onboarding_step or {})
|
|
required_onboarding_step = {
|
|
"profile_complete": True,
|
|
"workspace_create": True,
|
|
"workspace_invite": True,
|
|
"workspace_join": True,
|
|
}
|
|
next_onboarding_step = {**onboarding_step, **required_onboarding_step}
|
|
update_fields = []
|
|
|
|
if not profile.is_onboarded:
|
|
profile.is_onboarded = True
|
|
update_fields.append("is_onboarded")
|
|
|
|
if not profile.is_tour_completed:
|
|
profile.is_tour_completed = True
|
|
update_fields.append("is_tour_completed")
|
|
|
|
if profile.onboarding_step != next_onboarding_step:
|
|
profile.onboarding_step = next_onboarding_step
|
|
update_fields.append("onboarding_step")
|
|
|
|
if update_fields:
|
|
update_fields.append("updated_at")
|
|
profile.save(update_fields=list(dict.fromkeys(update_fields)))
|
|
|
|
|
|
def build_unique_username(email):
|
|
base_username = email.split("@", 1)[0].strip().lower() or "nodedc-user"
|
|
username = normalize_username(base_username)
|
|
|
|
if not User.objects.filter(username=username).exists():
|
|
return username
|
|
|
|
for _ in range(10):
|
|
candidate = f"{username}-{get_random_string(6).lower()}"
|
|
if not User.objects.filter(username=candidate).exists():
|
|
return candidate
|
|
|
|
return f"{username}-{get_random_string(12).lower()}"
|
|
|
|
|
|
def normalize_username(value):
|
|
normalized = "".join(char if char.isalnum() or char in {"_", "-", "."} else "-" for char in value)
|
|
normalized = normalized.strip("-._")
|
|
return normalized[:96] or "nodedc-user"
|
|
|
|
|
|
def sync_user_profile_from_claims(user, claims):
|
|
updated_fields = []
|
|
display_name = first_string_claim(claims, "name", "preferred_username")
|
|
given_name = first_string_claim(claims, "given_name")
|
|
family_name = first_string_claim(claims, "family_name")
|
|
avatar_url = normalize_nodedc_avatar_url(first_string_claim(claims, "picture", "avatar_url", "avatar"))
|
|
|
|
if display_name and user.display_name != display_name:
|
|
user.display_name = display_name
|
|
updated_fields.append("display_name")
|
|
|
|
if not given_name and display_name:
|
|
name_parts = display_name.split(" ", 1)
|
|
given_name = name_parts[0]
|
|
family_name = family_name or (name_parts[1] if len(name_parts) > 1 else "")
|
|
|
|
if given_name and user.first_name != given_name:
|
|
user.first_name = given_name
|
|
updated_fields.append("first_name")
|
|
|
|
if family_name is not None and user.last_name != family_name:
|
|
user.last_name = family_name
|
|
updated_fields.append("last_name")
|
|
|
|
if avatar_url and user.avatar != avatar_url:
|
|
user.avatar = avatar_url
|
|
updated_fields.append("avatar")
|
|
|
|
return updated_fields
|
|
|
|
|
|
def normalize_nodedc_avatar_url(value):
|
|
if not isinstance(value, str):
|
|
return ""
|
|
|
|
avatar_url = value.strip()
|
|
if not avatar_url:
|
|
return ""
|
|
|
|
if avatar_url.startswith(("http://", "https://", "data:")):
|
|
return avatar_url
|
|
|
|
if avatar_url.startswith(("/storage/", "/uploads/")):
|
|
launcher_origin = resolve_nodedc_launcher_origin()
|
|
return f"{launcher_origin}{avatar_url}" if launcher_origin else avatar_url
|
|
|
|
return avatar_url
|
|
|
|
|
|
def resolve_nodedc_launcher_origin():
|
|
explicit_origin = os.environ.get("PLANE_NODEDC_LAUNCHER_PUBLIC_URL", "").strip()
|
|
if explicit_origin:
|
|
return explicit_origin.rstrip("/")
|
|
|
|
for env_key in [
|
|
"PLANE_NODEDC_HANDOFF_URL",
|
|
"PLANE_NODEDC_ACCESS_CHECK_URL",
|
|
"PLANE_NODEDC_ACCESS_DENIED_REDIRECT_URL",
|
|
"PLANE_NODEDC_GLOBAL_LOGOUT_URL",
|
|
]:
|
|
configured_url = os.environ.get(env_key, "").strip()
|
|
if not configured_url:
|
|
continue
|
|
parsed_url = urlparse(configured_url)
|
|
if parsed_url.scheme and parsed_url.netloc:
|
|
return f"{parsed_url.scheme}://{parsed_url.netloc}"
|
|
|
|
return ""
|
|
|
|
|
|
def first_string_claim(claims, *keys):
|
|
for key in keys:
|
|
value = claims.get(key)
|
|
if isinstance(value, str) and value:
|
|
return value
|
|
return None
|
|
|
|
|
|
def hash_email(email):
|
|
normalized_email = str(email or "").strip().lower()
|
|
return hashlib.sha256(normalized_email.encode()).hexdigest()[:12] if normalized_email else ""
|
|
|
|
|
|
def hash_subject(subject):
|
|
normalized_subject = str(subject or "").strip()
|
|
return hashlib.sha256(normalized_subject.encode()).hexdigest()[:12] if normalized_subject else ""
|
|
|
|
|
|
def normalize_logout_guard_value(value):
|
|
return value.strip().lower() if isinstance(value, str) else ""
|
|
|
|
|
|
def get_logout_guard_cache_key(kind, value):
|
|
normalized_value = normalize_logout_guard_value(value)
|
|
return f"nodedc:logout-guard:{kind}:{normalized_value}" if normalized_value else ""
|
|
|
|
|
|
def get_logout_guard_cache_keys(claims):
|
|
keys = set()
|
|
|
|
subject_key = get_logout_guard_cache_key("subject", claims.get("sub"))
|
|
email_key = get_logout_guard_cache_key("email", claims.get("email"))
|
|
|
|
if subject_key:
|
|
keys.add(subject_key)
|
|
if email_key:
|
|
keys.add(email_key)
|
|
|
|
return sorted(keys)
|
|
|
|
|
|
def get_claim_auth_time(claims):
|
|
auth_time = claims.get("auth_time")
|
|
|
|
try:
|
|
return int(auth_time)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def get_logout_guard_time(claims):
|
|
guard_times = []
|
|
|
|
for cache_key in get_logout_guard_cache_keys(claims):
|
|
value = cache.get(cache_key)
|
|
|
|
try:
|
|
guard_times.append(int(value))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
return max(guard_times) if guard_times else 0
|
|
|
|
|
|
def clear_logout_guard(claims):
|
|
cache_keys = get_logout_guard_cache_keys(claims)
|
|
|
|
if cache_keys:
|
|
cache.delete_many(cache_keys)
|
|
|
|
|
|
def is_logout_guard_active(claims):
|
|
logout_guard_time = get_logout_guard_time(claims)
|
|
|
|
if not logout_guard_time:
|
|
return False
|
|
|
|
auth_time = get_claim_auth_time(claims)
|
|
|
|
if auth_time > logout_guard_time:
|
|
clear_logout_guard(claims)
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def oidc_error_redirect(base_url, next_path, error_code):
|
|
return HttpResponseRedirect(get_safe_redirect_url(base_url=base_url, next_path=next_path, params={"error": error_code}))
|
|
|
|
|
|
def oidc_login_redirect(base_url, next_path, prompt_login=False):
|
|
params = {}
|
|
|
|
if validate_next_path(next_path):
|
|
params["next_path"] = next_path
|
|
|
|
if prompt_login:
|
|
params["prompt"] = "login"
|
|
|
|
query_string = f"?{urlencode(params)}" if params else ""
|
|
|
|
return HttpResponseRedirect(f"{base_url.rstrip('/')}/auth/oidc/login/{query_string}")
|