admin / Syanpse-Vanguard

"""Extract the peer (client) certificate identity from the ASGI/TLS scope.

Uvicorn started with --ssl-cert-reqs 2 (CERT_REQUIRED) has already rejected any
client without a CA-signed cert by the time a request reaches the app. Here we
just read the CN so we can bind it to a known, non-revoked agent."""
from __future__ import annotations

from fastapi import Request


def client_ip(request: Request) -> str | None:
    """Best-effort source IP of the connecting agent. Honors the first hop of
    X-Forwarded-For when the app sits behind a trusted reverse proxy, otherwise
    uses the direct peer address."""
    xff = request.headers.get("x-forwarded-for")
    if xff:
        return xff.split(",")[0].strip()
    return request.client.host if request.client else None


def client_cn(request: Request) -> str | None:
    """Return the client cert Common Name, or None if no peer cert is present."""
    # Uvicorn surfaces the peer certificate under scope["extensions"]["tls"] or,
    # depending on version, the raw SSL object on the transport. Handle both.
    ext = request.scope.get("extensions") or {}
    tls = ext.get("tls") or {}
    peercert = tls.get("client_cert") or tls.get("peercert")
    if isinstance(peercert, dict):
        return _cn_from_parsed(peercert)

    transport = request.scope.get("transport")
    ssl_obj = getattr(transport, "get_extra_info", lambda _k: None)("ssl_object") \
        if transport else None
    if ssl_obj is not None:
        try:
            return _cn_from_parsed(ssl_obj.getpeercert())
        except Exception:
            return None
    return None


def _cn_from_parsed(peercert: dict | None) -> str | None:
    if not peercert:
        return None
    for rdn in peercert.get("subject", ()):
        for key, val in rdn:
            if key == "commonName":
                return val
    return None