|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Annotated
|
|
|
|
from fastapi import Header, Request
|
|
|
|
from devplacepy_services.base.errors import http_error
|
|
from devplacepy_services.base.schemas import InternalUser
|
|
|
|
_KEY_CACHE = ""
|
|
|
|
|
|
def internal_gateway_key() -> str:
|
|
global _KEY_CACHE
|
|
if _KEY_CACHE:
|
|
return _KEY_CACHE
|
|
env_key = os.environ.get("DEVPLACE_GATEWAY_INTERNAL_KEY", "").strip()
|
|
if env_key:
|
|
_KEY_CACHE = env_key
|
|
return env_key
|
|
try:
|
|
from devplacepy.db_client import internal_gateway_key as load_key
|
|
|
|
loaded = load_key().strip()
|
|
if loaded:
|
|
_KEY_CACHE = loaded
|
|
return loaded
|
|
except ImportError:
|
|
return ""
|
|
|
|
|
|
def set_internal_gateway_key(value: str) -> None:
|
|
global _KEY_CACHE
|
|
_KEY_CACHE = value.strip()
|
|
|
|
|
|
def validate_internal_key(presented: str | None) -> bool:
|
|
expected = internal_gateway_key()
|
|
if not expected:
|
|
return True
|
|
if not presented:
|
|
return False
|
|
return presented.strip() == expected
|
|
|
|
|
|
def require_internal_key(request: Request) -> None:
|
|
if validate_internal_key(request.headers.get("X-Internal-Key")):
|
|
return
|
|
raise http_error(401, "Unauthorized", "unauthorized")
|
|
|
|
|
|
async def internal_user(
|
|
request: Request,
|
|
x_authenticated_user: Annotated[str | None, Header()] = None,
|
|
) -> InternalUser:
|
|
require_internal_key(request)
|
|
uid = x_authenticated_user.strip() if x_authenticated_user else None
|
|
return InternalUser(uid=uid or None) |