Samuel AmarandClaude Haiku 4.5
77d7a50fa9
- FastAPI async gateway with httpx proxying to multiple upstreams - SQLite database with SQLAlchemy ORM - Admin console: manage services, users, API keys, endpoint access - Per-key, per-endpoint granular access control - OpenAPI document sync and caching (5-minute TTL) - Request/response logging with full transaction inspection - In-memory rate limiting (per-key, fixed-window) - Tiered log retention (7d payloads, 90d rows, incremental vacuum) - TLS verification toggle per service (for self-signed certificates) - Service connectivity validation with automatic endpoint refresh - Request browser with filters and deep-link inspection - Docker setup with persistent volume - Modal forms for create/edit flows Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
59 lines
1.6 KiB
Python
59 lines
1.6 KiB
Python
import hashlib
|
|
import hmac
|
|
import secrets
|
|
|
|
from itsdangerous import BadSignature, URLSafeTimedSerializer
|
|
|
|
from app import config
|
|
|
|
_serializer = URLSafeTimedSerializer(config.SECRET_KEY, salt="gw-session")
|
|
|
|
_PBKDF2_ITERATIONS = 600_000
|
|
|
|
|
|
# ---- password hashing (PBKDF2-SHA256) ----
|
|
|
|
def hash_password(password: str) -> str:
|
|
salt = secrets.token_hex(16)
|
|
digest = hashlib.pbkdf2_hmac(
|
|
"sha256", password.encode(), bytes.fromhex(salt), _PBKDF2_ITERATIONS
|
|
).hex()
|
|
return f"pbkdf2${_PBKDF2_ITERATIONS}${salt}${digest}"
|
|
|
|
|
|
def verify_password(password: str, stored: str) -> bool:
|
|
try:
|
|
_, iterations, salt, digest = stored.split("$")
|
|
computed = hashlib.pbkdf2_hmac(
|
|
"sha256", password.encode(), bytes.fromhex(salt), int(iterations)
|
|
).hex()
|
|
return hmac.compare_digest(computed, digest)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
# ---- API keys ----
|
|
|
|
def generate_api_key() -> tuple[str, str, str]:
|
|
"""Returns (plain_key, prefix, key_hash). Plain key is shown once."""
|
|
plain = "gw_" + secrets.token_urlsafe(32)
|
|
return plain, plain[:11], hash_api_key(plain)
|
|
|
|
|
|
def hash_api_key(plain: str) -> str:
|
|
return hashlib.sha256(plain.encode()).hexdigest()
|
|
|
|
|
|
# ---- session cookies ----
|
|
|
|
def create_session_token(user_id: int) -> str:
|
|
return _serializer.dumps({"uid": user_id})
|
|
|
|
|
|
def read_session_token(token: str) -> int | None:
|
|
try:
|
|
data = _serializer.loads(token, max_age=config.SESSION_MAX_AGE)
|
|
return int(data["uid"])
|
|
except (BadSignature, KeyError, ValueError, TypeError):
|
|
return None
|