Jun 27 MT-4
This commit is contained in:
@@ -0,0 +1,85 @@
|
||||
"""
|
||||
control/panel/impersonate.py
|
||||
-----------------------------
|
||||
Impersonation flow (MT-4).
|
||||
|
||||
How it works
|
||||
------------
|
||||
1. Superadmin clicks "Impersonate" on the panel → panel generates a short-lived
|
||||
signed token containing {tenant_id, expires_at, issued_by}.
|
||||
2. Panel redirects the superadmin's browser to
|
||||
https://<tenant-subdomain>/auth/impersonate?token=<token>
|
||||
(the main JQC app, not the panel).
|
||||
3. The main app's impersonation route (app/routes/auth.py) validates the token,
|
||||
sets a session key 'impersonating_tenant_id', and shows a banner.
|
||||
The tenancy middleware reads this key to bind the session to that tenant's DB
|
||||
even when multi-tenancy is enabled and the Host would resolve differently.
|
||||
4. "End impersonation" clears the key and redirects back to the panel.
|
||||
|
||||
Token format: HMAC-SHA256 signed JSON, base64url-encoded.
|
||||
|
||||
Environment
|
||||
-----------
|
||||
PANEL_IMPERSONATE_KEY — 32+ random hex bytes (separate from PANEL_SECRET_KEY).
|
||||
Generate: python -c "import secrets; print(secrets.token_hex(32))"
|
||||
Must be identical in /etc/jqc/control.env AND the main
|
||||
app's environment so both sides can verify tokens.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from base64 import urlsafe_b64decode, urlsafe_b64encode
|
||||
|
||||
_TTL_SECONDS = 60 # token is single-use; 60 s is more than enough for a redirect
|
||||
|
||||
|
||||
def _key() -> bytes:
|
||||
raw = os.environ.get('PANEL_IMPERSONATE_KEY', '')
|
||||
if not raw:
|
||||
raise RuntimeError(
|
||||
'PANEL_IMPERSONATE_KEY is not set. '
|
||||
'Generate: python -c "import secrets; print(secrets.token_hex(32))"'
|
||||
)
|
||||
return raw.encode()
|
||||
|
||||
|
||||
def _b64(data: bytes) -> str:
|
||||
return urlsafe_b64encode(data).rstrip(b'=').decode()
|
||||
|
||||
|
||||
def _unb64(s: str) -> bytes:
|
||||
pad = 4 - len(s) % 4
|
||||
return urlsafe_b64decode(s + '=' * (pad % 4))
|
||||
|
||||
|
||||
def generate_token(tenant_id: int, superadmin_id: int) -> str:
|
||||
"""Return a URL-safe signed token good for _TTL_SECONDS."""
|
||||
payload = json.dumps({
|
||||
'tid': tenant_id,
|
||||
'said': superadmin_id,
|
||||
'exp': int(time.time()) + _TTL_SECONDS,
|
||||
}, separators=(',', ':')).encode()
|
||||
sig = hmac.new(_key(), payload, hashlib.sha256).digest()
|
||||
return f'{_b64(payload)}.{_b64(sig)}'
|
||||
|
||||
|
||||
def validate_token(token: str) -> dict:
|
||||
"""Validate and decode a token. Returns payload dict or raises ValueError."""
|
||||
try:
|
||||
raw_payload, raw_sig = token.rsplit('.', 1)
|
||||
except ValueError:
|
||||
raise ValueError('Malformed token')
|
||||
|
||||
payload_bytes = _unb64(raw_payload)
|
||||
expected_sig = hmac.new(_key(), payload_bytes, hashlib.sha256).digest()
|
||||
if not hmac.compare_digest(expected_sig, _unb64(raw_sig)):
|
||||
raise ValueError('Invalid token signature')
|
||||
|
||||
payload = json.loads(payload_bytes)
|
||||
if time.time() > payload.get('exp', 0):
|
||||
raise ValueError('Token expired')
|
||||
|
||||
return payload
|
||||
Reference in New Issue
Block a user