86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
"""
|
|
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
|