73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
"""Security and token generation for standalone draft review."""
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
|
|
def _b64url_encode(data: bytes) -> str:
|
|
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
|
|
|
|
|
|
def _b64url_decode(s: str) -> bytes:
|
|
padding = 4 - (len(s) % 4)
|
|
if padding != 4:
|
|
s += "=" * padding
|
|
return base64.urlsafe_b64decode(s.encode("ascii"))
|
|
|
|
|
|
def create_draft_token(
|
|
draft_id: str | UUID,
|
|
chat_id: int | None = None,
|
|
user_id: str | None = None,
|
|
secret: str = "procurement-draft-secret-change-in-prod-v7",
|
|
expires_in: int = 7200,
|
|
) -> str:
|
|
"""Generate an HMAC-SHA256 signed token for secure draft access."""
|
|
now = int(time.time())
|
|
payload = {
|
|
"draft_id": str(draft_id),
|
|
"chat_id": chat_id,
|
|
"user_id": str(user_id) if user_id else None,
|
|
"iat": now,
|
|
"exp": now + expires_in,
|
|
}
|
|
payload_bytes = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
|
payload_b64 = _b64url_encode(payload_bytes)
|
|
|
|
sig = hmac.new(secret.encode("utf-8"), payload_b64.encode("ascii"), hashlib.sha256).digest()
|
|
sig_b64 = _b64url_encode(sig)
|
|
|
|
return f"{payload_b64}.{sig_b64}"
|
|
|
|
|
|
def verify_draft_token(
|
|
token: str,
|
|
secret: str = "procurement-draft-secret-change-in-prod-v7",
|
|
) -> dict[str, Any] | None:
|
|
"""Verify signature and expiration of a draft token. Return payload or None."""
|
|
if not token or "." not in token:
|
|
return None
|
|
try:
|
|
payload_b64, sig_b64 = token.split(".", 1)
|
|
expected_sig = hmac.new(secret.encode("utf-8"), payload_b64.encode("ascii"), hashlib.sha256).digest()
|
|
actual_sig = _b64url_decode(sig_b64)
|
|
|
|
if not hmac.compare_digest(expected_sig, actual_sig):
|
|
return None
|
|
|
|
payload_bytes = _b64url_decode(payload_b64)
|
|
payload = json.loads(payload_bytes.decode("utf-8"))
|
|
|
|
# Check expiration
|
|
exp = payload.get("exp")
|
|
if exp is not None and int(exp) < int(time.time()):
|
|
return None
|
|
|
|
return payload
|
|
except Exception:
|
|
return None
|