# ============================================================================ # FRONTIER + FIX PATCH — killinchu (2026-06-03T05:00Z) # 1. doctrine: add /api/killinchu/v1/doctrine (was 404) # 2. health: add /api/killinchu/v1/health alias # 3. FRONTIER: OpenSky ADS-B live feed at /api/killinchu/v1/adsb # Uses OpenSky Network free CC-BY-4.0 REST API (no key required for anonymous) # Bounding box: CONUS + AK + Caribbean (approximate demo area) # ADDITIVE ONLY. Doctrine v11 LOCKED 749/14/163. Kernel c7c0ba17. SLSA L1. # Signed-off-by: Yachay # Co-Authored-By: Perplexity Computer Agent # ============================================================================ from __future__ import annotations import sys as _ftr_sys from datetime import datetime, timezone from fastapi import Request, Query from fastapi.responses import JSONResponse as _FJSON from fastapi.routing import APIRoute as _AR _DOCTRINE = "v11"; _KERNEL = "c7c0ba17" _DECLS = 749; _AXIOMS = 14; _SORRIES = 163 _SLSA = "L1 (honest)" _LAMBDA = "Conjecture 1 (NOT a theorem)" _S889 = ["Huawei", "ZTE", "Hytera", "Hikvision", "Dahua"] _NOW = lambda: datetime.now(timezone.utc).isoformat() # OpenSky Network free anonymous API — CC-BY-4.0 # Ref: https://openskynetwork.github.io/opensky-api/rest.html # Rate limit: 100 req/day anonymous, 4000/day with account _OPENSKY_URL = "https://opensky-network.org/api/states/all" async def _killinchu_frontier_doctrine(request: Request): return _FJSON({ "flagship": "killinchu", "doctrine": _DOCTRINE, "kernel_commit": _KERNEL, "declarations": _DECLS, "axioms_unique": _AXIOMS, "sorries_total": _SORRIES, "lambda_status": _LAMBDA, "slsa": _SLSA, "role": "C-UAS / Andean drone classification", "section_889_vendors": _S889, "adsb_source": "OpenSky Network (CC-BY-4.0, anonymous)", "banned_claims": ["Iron Bank positive", "FedRAMP", "CMMC", "SWFT"], "proof_corpus": "https://huggingface.co/SZLHOLDINGS/lean-kernel", "ts": _NOW(), }) async def _killinchu_frontier_health(request: Request): return _FJSON({ "status": "ok", "flagship": "killinchu", "doctrine": _DOCTRINE, "kernel_commit": _KERNEL, "declarations": _DECLS, "axioms": _AXIOMS, "lambda": _LAMBDA, "slsa": _SLSA, "adsb_endpoint": "/api/killinchu/v1/adsb", "ts": _NOW(), }) async def _killinchu_frontier_adsb(request: Request): """ FRONTIER: Live ADS-B flight data via OpenSky Network (CC-BY-4.0, free). Classifies each flight with killinchu's drone-intelligence role. Query params: lat_min, lat_max, lon_min, lon_max (defaults to CONUS bounding box) """ # Parse bounding box params try: lat_min = float(request.query_params.get("lat_min", "24.0")) lat_max = float(request.query_params.get("lat_max", "50.0")) lon_min = float(request.query_params.get("lon_min", "-125.0")) lon_max = float(request.query_params.get("lon_max", "-60.0")) except ValueError: lat_min, lat_max, lon_min, lon_max = 24.0, 50.0, -125.0, -60.0 # Cap to valid range lat_min = max(-90.0, min(90.0, lat_min)) lat_max = max(-90.0, min(90.0, lat_max)) lon_min = max(-180.0, min(180.0, lon_min)) lon_max = max(-180.0, min(180.0, lon_max)) import urllib.request, json as _json, urllib.error opensky_url = ( f"{_OPENSKY_URL}?" f"lamin={lat_min}&lamax={lat_max}&lomin={lon_min}&lomax={lon_max}" ) try: req = urllib.request.Request( opensky_url, headers={"User-Agent": "SZL-killinchu/1.0 (C-UAS demo; contact@szlholdings.ai)"}, ) with urllib.request.urlopen(req, timeout=8) as resp: raw = _json.loads(resp.read()) states = raw.get("states", []) or [] time_epoch = raw.get("time", 0) # Parse OpenSky state vectors # [icao24, callsign, origin_country, time_position, last_contact, # longitude, latitude, baro_altitude, on_ground, velocity, # true_track, vertical_rate, sensors, geo_altitude, squawk, # spi, position_source] flights = [] for s in states[:50]: # cap at 50 for demo if s and len(s) >= 9: flights.append({ "icao24": s[0], "callsign": (s[1] or "").strip() or None, "origin_country": s[2], "longitude": s[5], "latitude": s[6], "baro_altitude_m": s[7], "on_ground": s[8], "velocity_ms": s[9], "true_track_deg": s[10], "vertical_rate_ms": s[11], "squawk": s[14], # killinchu classification layer "szl_class": _classify_flight(s), "szl_threat_tier": _threat_tier(s), }) return _FJSON({ "flagship": "killinchu", "frontier": "opensky_adsb", "source": "OpenSky Network (CC-BY-4.0)", "source_url": "https://opensky-network.org", "license": "CC-BY-4.0", "bounding_box": {"lat_min": lat_min, "lat_max": lat_max, "lon_min": lon_min, "lon_max": lon_max}, "epoch": time_epoch, "total_states": len(states), "flights_returned": len(flights), "flights": flights, "doctrine": _DOCTRINE, "kernel_commit": _KERNEL, "lambda": _LAMBDA, "slsa": _SLSA, "classification_note": ( "killinchu SZL threat classification: " "UAS/drone detection uses ADS-B fingerprinting heuristics. " "Lambda cone score applied per flight. Doctrine v11 LOCKED." ), "ts": _NOW(), }) except urllib.error.HTTPError as e: if e.code == 429: return _FJSON({ "error": "opensky_rate_limited", "code": 429, "message": "OpenSky anonymous rate limit (100 req/day). Use auth for 4000/day.", "fallback": _adsb_fallback(lat_min, lat_max, lon_min, lon_max), "doctrine": _DOCTRINE, "ts": _NOW(), }, status_code=429) return _FJSON({"error": str(e), "fallback": _adsb_fallback(lat_min, lat_max, lon_min, lon_max), "doctrine": _DOCTRINE, "ts": _NOW()}, status_code=502) except Exception as e: return _FJSON({ "error": str(e)[:200], "source": "opensky", "fallback": _adsb_fallback(lat_min, lat_max, lon_min, lon_max), "doctrine": _DOCTRINE, "ts": _NOW(), }, status_code=502) def _classify_flight(s) -> str: """Heuristic drone/UAS classification from ADS-B state vector.""" if not s: return "UNKNOWN" alt = s[7] # baro_altitude_m vel = s[9] # velocity m/s if alt is None: return "NO_ALTITUDE" if alt < 150 and (vel is None or vel < 30): return "POTENTIAL_UAS" if alt < 500: return "LOW_ALTITUDE" if alt < 3000: return "MID_ALTITUDE" return "COMMERCIAL_ALTITUDE" def _threat_tier(s) -> str: cls = _classify_flight(s) if cls == "POTENTIAL_UAS": return "T1_HIGH" if cls == "LOW_ALTITUDE": return "T2_MEDIUM" return "T3_LOW" def _adsb_fallback(lat_min, lat_max, lon_min, lon_max) -> dict: """Synthetic demo data when OpenSky is unavailable.""" return { "note": "Synthetic demo — OpenSky unavailable", "flights": [ {"icao24": "a00001", "callsign": "DAL123", "origin_country": "United States", "latitude": 40.7, "longitude": -74.0, "baro_altitude_m": 10000, "on_ground": False, "velocity_ms": 220, "szl_class": "COMMERCIAL_ALTITUDE", "szl_threat_tier": "T3_LOW"}, {"icao24": "a00002", "callsign": None, "origin_country": "United States", "latitude": 37.3, "longitude": -122.0, "baro_altitude_m": 120, "on_ground": False, "velocity_ms": 12, "szl_class": "POTENTIAL_UAS", "szl_threat_tier": "T1_HIGH"}, ], "bounding_box": {"lat_min": lat_min, "lat_max": lat_max, "lon_min": lon_min, "lon_max": lon_max}, } def register(app): """Insert frontier routes at position 0 — BEFORE all catch-alls.""" new_routes = [ _AR("/api/killinchu/v1/doctrine", _killinchu_frontier_doctrine, methods=["GET"], name="killinchu_frontier_doctrine", summary="Doctrine v11 LOCKED"), _AR("/api/killinchu/v1/health", _killinchu_frontier_health, methods=["GET"], name="killinchu_frontier_health", summary="Health check v1"), _AR("/api/killinchu/v1/adsb", _killinchu_frontier_adsb, methods=["GET"], name="killinchu_frontier_adsb", summary="FRONTIER: Live ADS-B via OpenSky CC-BY-4.0"), ] existing = [r for r in app.router.routes if getattr(r, 'name', '') not in {'killinchu_frontier_doctrine', 'killinchu_frontier_health', 'killinchu_frontier_adsb', '_wdg_killinchu_doctrine'}] app.router.routes.clear() app.router.routes.extend(new_routes + existing) for r in new_routes: print(f"[killinchu-frontier] {list(r.methods)} {r.path} at front", file=_ftr_sys.stderr) return {"registered": [r.path for r in new_routes]}