49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
import hmac
|
|
from typing import Optional
|
|
from flask import Request, current_app, g, request as flask_request
|
|
import jwt
|
|
|
|
API_KEY_HEADER = "X-Internal-Xcloudify-API"
|
|
|
|
|
|
def get_request_user_id(request: Request) -> Optional[str]:
|
|
"""
|
|
Extract the acting user_id from an Authorization Bearer JWT (HS256).
|
|
|
|
Returns:
|
|
str | None: The 'sub' claim from the JWT if present and valid; otherwise None.
|
|
"""
|
|
auth_header = request.headers.get("Authorization", "")
|
|
if not auth_header or not auth_header.startswith("Bearer "):
|
|
return None
|
|
|
|
token = auth_header.split(" ", 1)[1].strip()
|
|
if not token:
|
|
return None
|
|
|
|
secret = current_app.config.get("JWT_SECRET_KEY") or "your-very-secret-key"
|
|
try:
|
|
payload = jwt.decode(token, secret, algorithms=["HS256"])
|
|
sub = payload.get("sub")
|
|
return str(sub) if sub is not None else None
|
|
except Exception:
|
|
# On any JWT decode/validation error, do not block the request; just omit user_id
|
|
return None
|
|
|
|
|
|
def _accepted_keys() -> list:
|
|
raw = current_app.config.get("XCLOUDIFY_API_KEY") or ""
|
|
return [k.strip() for k in raw.split(",") if k.strip()]
|
|
|
|
|
|
def authenticate_request():
|
|
g.is_service = False
|
|
presented = flask_request.headers.get(API_KEY_HEADER)
|
|
if not presented:
|
|
return
|
|
presented_bytes = presented.encode("utf-8")
|
|
for key in _accepted_keys():
|
|
if hmac.compare_digest(presented_bytes, key.encode("utf-8")):
|
|
g.is_service = True
|
|
return
|