|
# retoor <retoor@molodetz.nl>
|
|
|
|
from typing import Optional
|
|
from urllib.parse import urlsplit
|
|
from fastapi import Request, HTTPException, status
|
|
from devplacepy.utils.auth import get_current_user, _request_has_auth
|
|
|
|
|
|
def safe_next(value, default="/feed"):
|
|
if (
|
|
value
|
|
and value.startswith("/")
|
|
and not value.startswith(("//", "/\\"))
|
|
and "\\" not in value
|
|
and "\n" not in value
|
|
and "\r" not in value
|
|
):
|
|
return value
|
|
return default
|
|
|
|
|
|
def redirect_back(request: Request, default: str = "/feed") -> str:
|
|
referer = request.headers.get("referer") or ""
|
|
parts = urlsplit(referer)
|
|
if parts.scheme and parts.netloc:
|
|
if parts.netloc == urlsplit(str(request.base_url)).netloc:
|
|
referer = parts.path + (f"?{parts.query}" if parts.query else "")
|
|
else:
|
|
referer = ""
|
|
return safe_next(referer, default)
|
|
|
|
|
|
def cookie_secure(request: Request) -> bool:
|
|
forwarded = request.headers.get("x-forwarded-proto", "")
|
|
if forwarded:
|
|
return forwarded.split(",")[0].strip().lower() == "https"
|
|
return request.url.scheme == "https"
|
|
|
|
|
|
def client_ip(request, default: Optional[str] = "unknown") -> Optional[str]:
|
|
headers = getattr(request, "headers", None)
|
|
if headers is not None:
|
|
real = headers.get("x-real-ip")
|
|
if real and real.strip():
|
|
return real.strip()
|
|
forwarded = headers.get("x-forwarded-for")
|
|
if forwarded:
|
|
first = forwarded.split(",")[0].strip()
|
|
if first:
|
|
return first
|
|
client = getattr(request, "client", None)
|
|
if client is not None:
|
|
return client.host
|
|
return default
|
|
|
|
|
|
def require_user(request: Request):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
from devplacepy.responses import wants_json
|
|
from devplacepy.services.audit import record as audit
|
|
|
|
audit.record(
|
|
request,
|
|
"security.authz.denied",
|
|
user=None,
|
|
result="denied",
|
|
summary=f"guest denied access to {request.url.path}",
|
|
)
|
|
if _request_has_auth(request) or wants_json(request):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials"
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/auth/login"}
|
|
)
|
|
return user
|
|
|
|
|
|
def is_admin(user: dict | None) -> bool:
|
|
return bool(user) and user.get("role") == "Admin"
|
|
|
|
|
|
def is_primary_admin(user: dict | None) -> bool:
|
|
from devplacepy.database import get_primary_admin_uid
|
|
|
|
return is_admin(user) and user.get("uid") == get_primary_admin_uid()
|
|
|
|
|
|
def require_admin(request: Request):
|
|
user = require_user(request)
|
|
if not is_admin(user):
|
|
from devplacepy.responses import wants_json
|
|
from devplacepy.services.audit import record as audit
|
|
|
|
audit.record(
|
|
request,
|
|
"security.authz.denied",
|
|
user=user,
|
|
result="denied",
|
|
summary=f"non-admin denied access to {request.url.path}",
|
|
)
|
|
if wants_json(request):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Administrator access required",
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/feed"}
|
|
)
|
|
return user
|
|
|
|
|
|
def not_found(detail: str = "Not found") -> HTTPException:
|
|
return HTTPException(status_code=404, detail=detail)
|
|
|
|
|
|
def require_user_api(request: Request):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
from devplacepy.services.audit import record as audit
|
|
|
|
audit.record(
|
|
request,
|
|
"security.authz.denied",
|
|
user=None,
|
|
result="denied",
|
|
summary=f"guest denied access to {request.url.path}",
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required"
|
|
)
|
|
return user
|