feat(core): add REALITY target scanner

app/models/reality_scan.py

@@ -0,0 +1,38 @@
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class RealityScanRequest(BaseModel):
+    target: str = Field(min_length=1, max_length=253, description="host or host:port to probe (port defaults to 443)")
+    sni: str | None = Field(default=None, max_length=253, description="Override the SNI sent during the handshake")
+    timeout: float | None = Field(default=None, ge=1, le=20, description="Per-probe timeout in seconds (1-20, default 10)")
+
+
+class RealityScanResult(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    target: str
+    host: str
+    ip: str | None = None
+    port: int
+    sni: str | None = None
+
+    feasible: bool
+    tls13: bool
+    tls_version: str | None = None
+    h2: bool
+    alpn: str | None = None
+
+    x25519: bool | None = None
+    post_quantum: bool | None = None
+    curve: str | None = None
+
+    h3: bool = False
+
+    cert_valid: bool
+    cert_subject: str | None = None
+    cert_issuer: str | None = None
+    not_after: str | None = None
+    server_names: list[str] = Field(default_factory=list)
+
+    latency_ms: int | None = None
+    reason: str | None = None

app/operation/core.py

@@ -25,8 +25,10 @@
     CoresSimpleResponse,
     RemoveCoresResponse,
 )
+from app.models.reality_scan import RealityScanRequest, RealityScanResult
 from app.operation import BaseOperation
 from app.utils.logger import get_logger
+from app.utils.reality_scan import RealityScanError, scan_reality_target
 
 logger = get_logger("core-operation")
 
@@ -36,6 +38,16 @@ async def _refresh_hosts_from_db(self, db: AsyncSession) -> None:
         db_hosts = await get_hosts(db=db)
         await host_manager.add_hosts(db, db_hosts)
 
+    async def scan_reality_target(self, request: RealityScanRequest) -> RealityScanResult:
+        try:
+            result = await scan_reality_target(target=request.target, sni=request.sni, timeout=request.timeout)
+        except RealityScanError as e:
+            await self.raise_error(message=str(e), code=400)
+        except Exception as e:
+            logger.warning("reality scan failed for %r: %s", request.target, e)
+            await self.raise_error(message=f"Reality scan failed: {e}", code=502)
+        return RealityScanResult.model_validate(result)
+
     async def create_core(self, db: AsyncSession, new_core: CoreCreate, admin: AdminDetails) -> CoreResponse:
         try:
             validated_core = core_manager.validate_core(

app/routers/core.py

@@ -10,6 +10,7 @@
     CoresSimpleResponse,
     RemoveCoresResponse,
 )
+from app.models.reality_scan import RealityScanRequest, RealityScanResult
 from app.operation import OperatorType
 from app.operation.core import CoreOperation
 from app.operation.node import NodeOperation
@@ -33,6 +34,14 @@ async def create_core_config(
     return await core_operator.create_core(db, new_core, admin)
 
 
+@router.post("/reality-scan", response_model=RealityScanResult)
+async def scan_reality_target(
+    request: RealityScanRequest,
+    _: AdminDetails = Depends(require_permission("cores", "read")),
+):
+    return await core_operator.scan_reality_target(request)
+
+
 @router.get("/{core_id}", response_model=CoreResponse)
 async def get_core_config(
     core_id: int, _: AdminDetails = Depends(require_permission("cores", "read")), db: AsyncSession = Depends(get_db)