.env.example

@@ -31,6 +31,20 @@ NOMINATIM_URL=https://nominatim.openstreetmap.org
 NOMINATIM_MIN_INTERVAL=1.1
 SNAPSHOT_TTL_SECONDS=300
 
+# ── masscan active camera discovery (host-level systemd service, NOT compose) ─
+# Continuous rolling sweep for open RTSP port 554 across a range. Runs on the
+# Pi host via deploy/osint-masscan.service (needs root + raw sockets). Results
+# land in the same `cameras` table as the scraper (discovery_source=masscan).
+# NOTE: 200 pps is the residential-safe default. 1k/10k pps saturated a home
+# uplink. A full 0.0.0.0/0 sweep at 200 pps takes ~8 months (rolling).
+MASSCAN_RANGE=0.0.0.0/0
+MASSCAN_PORTS=554
+MASSCAN_RATE=200
+MASSCAN_RETRIES=1
+MASSCAN_WAIT=0
+MASSCAN_EXCLUDEFILE=/etc/osint-dashboard/masscan-excludes.txt
+MASSCAN_FLUSH_EVERY=250
+
 # ── NASA FIRMS (active fire / hotspot ingest) ──────────────────────────────
 # MAP_KEY is FREE — get one at https://firms.modaps.eosdis.nasa.gov/api/map_key_info/
 # (1-minute signup, no payment). Leave blank to keep fire ingest idle.

alembic/versions/011_geofence_alerts_fence.py

@@ -1,24 +0,0 @@
-"""geofence_alerts (geofence_id, created_at DESC) for fence-scoped hit log
-
-Revision ID: 011_geofence_alerts_fence
-Revises: 010_bbox_gist
-Create Date: 2026-09-01
-"""
-
-from alembic import op
-
-revision = "011_geofence_alerts_fence"
-down_revision = "010_bbox_gist"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    op.execute(
-        "CREATE INDEX IF NOT EXISTS ix_geofence_alerts_fence_created "
-        "ON geofence_alerts (geofence_id, created_at DESC)"
-    )
-
-
-def downgrade() -> None:
-    op.execute("DROP INDEX IF EXISTS ix_geofence_alerts_fence_created")

app/bg_jobs.py

@@ -1,19 +1,50 @@
-"""Background ffmpeg — never block a FastAPI request on a frame grab.
+"""Background masscan / ffmpeg — never block a FastAPI request on a scan.
 
-ffmpeg frame grabs are scheduled with asyncio.create_task and shared per URL.
+masscan is capped at 200 pps (home uplink saturates at 1k+). ffmpeg frame
+grabs are scheduled with asyncio.create_task and shared per URL.
 """
 
 from __future__ import annotations
 
 import asyncio
+import logging
 import shutil
 from cachetools import TTLCache
 
+logger = logging.getLogger("osint.bg_jobs")
+
+MASSCAN_PPS_CAP = 200
+
+_masscan_task: asyncio.Task | None = None
 _ffmpeg_cache: TTLCache = TTLCache(maxsize=100, ttl=300)
 _ffmpeg_tasks: dict[str, asyncio.Task] = {}
 _FFMPEG = shutil.which("ffmpeg")
 
 
+def schedule_masscan_pass() -> bool:
+    """Kick one capped masscan pass. Returns False if a pass is already running."""
+    global _masscan_task
+    if _masscan_task is not None and not _masscan_task.done():
+        return False
+    _masscan_task = asyncio.create_task(_run_masscan_capped())
+    return True
+
+
+async def _run_masscan_capped() -> None:
+    import masscan_config as cfg
+    from run_masscan_service import _verify_excludefile, run_pass
+
+    orig = cfg.MASSCAN_RATE
+    if orig > MASSCAN_PPS_CAP:
+        logger.warning("capping masscan rate %s pps -> %s", orig, MASSCAN_PPS_CAP)
+        cfg.MASSCAN_RATE = MASSCAN_PPS_CAP
+    try:
+        _verify_excludefile()
+        await run_pass()
+    finally:
+        cfg.MASSCAN_RATE = orig
+
+
 def cached_ffmpeg_jpeg(url: str) -> bytes | None:
     return _ffmpeg_cache.get(url)
 

app/camera_preview.py

@@ -1,7 +1,7 @@
 """Resolve a browser-renderable preview for a camera.
 
 HTTP/MJPEG cameras already expose a snapshot_url the existing proxy can
-stream. Some scraper sources store `rtsp://` URLs with no snapshot_url, so
+stream. masscan finds are stored as `rtsp://IP/` with no snapshot_url, so
 the map popup used to skip the <img> entirely and the leftover source link
 handed the browser an rtsp:// URL (which opens VLC).
 
@@ -42,6 +42,16 @@ _HTTP_PATHS = (
     "/tmpfs/auto.jpg",
 )
 
+# Browser-playable MJPEG paths the /stream proxy can pass through.
+_MJPEG_PATHS = (
+    "/mjpg/video.mjpg",
+    "/video.mjpg",
+    "/cgi-bin/mjpg/video.cgi",
+    "/axis-cgi/mjpg/video.cgi",
+    "/nphMotionJpeg",
+    "/mjpeg.cgi",
+)
+
 _FFMPEG = shutil.which("ffmpeg")
 
 
@@ -77,6 +87,55 @@ async def _http_get_image(url: str, timeout: float = 2.5) -> bytes | None:
         return None
 
 
+async def _http_feed_url(url: str, timeout: float = 2.5) -> str | None:
+    """Return url if it looks like an unauthenticated image/MJPEG feed."""
+    try:
+        async with httpx.AsyncClient(
+            timeout=timeout, follow_redirects=True,
+            headers={"User-Agent": USER_AGENT},
+        ) as c:
+            async with c.stream("GET", url) as r:
+                if r.status_code != 200:
+                    return None
+                ctype = (r.headers.get("content-type") or "").lower()
+                if "html" in ctype or ctype.startswith("text/"):
+                    return None
+                if any(x in ctype for x in ("image/", "multipart", "mjpeg", "octet-stream")):
+                    # Read a little to reject empty/error bodies.
+                    chunk = b""
+                    async for b in r.aiter_bytes():
+                        chunk += b
+                        if len(chunk) >= 64:
+                            break
+                    if len(chunk) < 64:
+                        return None
+                    if b"html" in chunk[:64].lower():
+                        return None
+                    return url
+    except Exception:  # noqa: BLE001
+        return None
+    return None
+
+
+async def probe_public_feed(host: str) -> str | None:
+    """Unauthenticated HTTP still or MJPEG URL for this host, or None.
+
+    Used at masscan ingest time so dead RTSP-only hosts never hit the map.
+    No credentials, no RTSP path-walking (too slow / rarely public).
+    """
+    urls = [f"http://{host}{p}" for p in _HTTP_PATHS]
+    urls.append(f"http://{host}:8080/shot.jpg")
+    urls.extend(f"http://{host}{p}" for p in _MJPEG_PATHS)
+    results = await asyncio.gather(
+        *(_http_feed_url(u) for u in urls),
+        return_exceptions=True,
+    )
+    for url, hit in zip(urls, results):
+        if isinstance(hit, str) and hit:
+            return hit
+    return None
+
+
 async def ffmpeg_snapshot(url: str, timeout: float = 8.0) -> bytes | None:
     """Grab a single JPEG frame from an RTSP URL. None if ffmpeg missing/fails.
 

app/geofence.py

@@ -319,154 +319,3 @@ async def record_and_notify(
         except Exception:
             pass
     return sent
-
-
-async def list_alerts(
-    *,
-    geofence_id: str | None = None,
-    since: datetime | None = None,
-    until: datetime | None = None,
-    source_kind: str | None = None,
-    limit: int = 100,
-) -> list[dict]:
-    """Filterable hit log. Empty list if the DB is down — never raises."""
-    where = ["TRUE"]
-    params: dict[str, Any] = {"limit": int(limit)}
-    if geofence_id:
-        where.append("geofence_id = CAST(:geofence_id AS uuid)")
-        params["geofence_id"] = geofence_id
-    if since is not None:
-        where.append("created_at >= :since")
-        params["since"] = since
-    if until is not None:
-        where.append("created_at <= :until")
-        params["until"] = until
-    if source_kind:
-        where.append("source_kind = :source_kind")
-        params["source_kind"] = source_kind
-    sql = f"""
-        SELECT id::text, geofence_id::text, source_kind, entity_id,
-               lat, lon, payload, created_at
-        FROM geofence_alerts
-        WHERE {' AND '.join(where)}
-        ORDER BY created_at DESC
-        LIMIT :limit
-    """
-    try:
-        async with async_session() as session:
-            rows = (await session.execute(text(sql), params)).mappings().all()
-        out = []
-        for r in rows:
-            item = dict(r)
-            if item.get("created_at") is not None:
-                item["created_at"] = item["created_at"].isoformat()
-            out.append(item)
-        return out
-    except Exception:
-        return []
-
-
-async def get_geofence(gid: str) -> dict | None:
-    current = next((f for f in _cache if f["id"] == gid), None)
-    if current is not None:
-        return current
-    try:
-        await refresh_cache()
-    except Exception:
-        return None
-    return next((f for f in _cache if f["id"] == gid), None)
-
-
-def _marker_from_track(row) -> dict:
-    from live_layers import to_marker
-
-    extra = {"bucket": row["bucket"].isoformat() if row.get("bucket") else None, "dvr": True}
-    return to_marker(
-        row["id"], row["lat"], row["lon"],
-        heading=row.get("heading"), speed=row.get("speed"),
-        label=row.get("label") or row["id"],
-        extra=extra,
-    )
-
-
-async def _cagg_inside(gid: str, kind: str, bucket: datetime, limit: int = 2000) -> list[dict]:
-    table = "aircraft_tracks_1min" if kind == "aircraft" else "vessel_tracks_1min"
-    id_col = "hex" if kind == "aircraft" else "mmsi"
-    sql = f"""
-        SELECT {id_col} AS id, lat, lon, heading, speed, label, bucket
-        FROM {table}
-        WHERE bucket = :bucket
-          AND ST_Intersects(
-                (SELECT geom FROM geofences WHERE id = CAST(:gid AS uuid)),
-                ST_SetSRID(ST_MakePoint(lon, lat), 4326)
-              )
-        LIMIT :limit
-    """
-    try:
-        async with async_session() as session:
-            rows = (await session.execute(
-                text(sql), {"bucket": bucket, "gid": gid, "limit": limit},
-            )).mappings().all()
-        return [
-            _marker_from_track(r)
-            for r in rows
-            if r["lat"] is not None and r["lon"] is not None
-        ]
-    except Exception:
-        return []
-
-
-async def _fires_inside(gid: str, ts: datetime, limit: int = 2000) -> list[dict]:
-    from tracks import minute_bucket
-
-    bucket = minute_bucket(ts)
-    t1 = bucket + timedelta(minutes=1)
-    sql = """
-        SELECT latitude, longitude, brightness, confidence, acq_time, satellite,
-               instrument, bright_ti5, frp, daynight
-        FROM fires
-        WHERE acq_time >= :t0 AND acq_time < :t1
-          AND ST_Intersects(
-                (SELECT geom FROM geofences WHERE id = CAST(:gid AS uuid)),
-                ST_SetSRID(ST_MakePoint(longitude, latitude), 4326)
-              )
-        LIMIT :limit
-    """
-    try:
-        async with async_session() as session:
-            rows = (await session.execute(
-                text(sql),
-                {"t0": bucket, "t1": t1, "gid": gid, "limit": limit},
-            )).mappings().all()
-        out = []
-        for r in rows:
-            item = dict(r)
-            if item.get("acq_time") is not None:
-                item["acq_time"] = item["acq_time"].isoformat()
-            out.append(item)
-        return out
-    except Exception:
-        return []
-
-
-async def snapshot_at(gid: str, ts: datetime) -> dict | None:
-    """Positions inside the fence at time T. None if the fence is missing.
-
-    Does not persist or notify. Empty lists if track/fire queries fail.
-    """
-    fence = await get_geofence(gid)
-    if fence is None:
-        return None
-    from tracks import minute_bucket
-
-    bucket = minute_bucket(ts)
-    aircraft = await _cagg_inside(gid, "aircraft", bucket)
-    vessels = await _cagg_inside(gid, "vessel", bucket)
-    fires = await _fires_inside(gid, ts)
-    return {
-        "geofence_id": gid,
-        "timestamp": ts.isoformat(),
-        "aircraft": aircraft,
-        "vessels": vessels,
-        "fires": fires,
-    }

app/main.py

@@ -21,12 +21,11 @@ from datetime import datetime, timedelta, timezone
 from decimal import Decimal
 from pathlib import Path
 from typing import NoReturn
-from urllib.parse import urlparse
 from uuid import UUID
 
 import httpx
 import structlog
-from fastapi import FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect
+from fastapi import BackgroundTasks, FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect
 from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
@@ -44,7 +43,7 @@ from schemas import (
     DashboardSummary, EntityCreate, EntityKind, EntityOut,
     EventCreate, EventOut, FireOut, NewsArticleOut, NewsMapItemOut,
     NewsSummaryOut, NewsTickerItemOut,
-    FeedSourceCreate, FeedSourceOut, FeedSourceUpdate,
+    FeedSourceCreate, FeedSourceOut,
     KeyOut, KeyValueIn,
     NewsModelsOut, SettingsIn, SettingsOut,
     SearchResult, SentimentSummary, SourceType,
@@ -52,8 +51,7 @@ from schemas import (
     GeofenceCreate, GeofenceUpdate,
 )
 from ingestor import ingest_event, fetch_and_process
-from camera_scraper import is_public_url
-from sources import GDELT_API, ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals
+from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals
 from fire_sources import ingest_fires
 from keystore import KeyFormatError, delete_key, list_keys, set_key
 from settings_store import SettingsError, get_app_settings, list_models, set_summary_model
@@ -368,22 +366,20 @@ async def create_source(payload: FeedSourceCreate):
 
 
 @app.patch("/api/sources/{source_id}")
-async def update_source(source_id: UUID, payload: FeedSourceUpdate):
-    """Update a feed source (name/url/config/enabled only)."""
-    values = payload.model_dump(exclude_unset=True)
+async def update_source(source_id: UUID, payload: dict):
+    """Update a feed source (e.g., toggle enabled)."""
     async with async_session() as session:
         row = (await session.execute(
             select(feed_sources).where(feed_sources.c.id == source_id)
         )).mappings().one_or_none()
         if not row:
             raise HTTPException(404, "Source not found")
-        if values:
-            await session.execute(
-                feed_sources.update()
-                .where(feed_sources.c.id == source_id)
-                .values(**values)
-            )
-            await session.commit()
+        await session.execute(
+            feed_sources.update()
+            .where(feed_sources.c.id == source_id)
+            .values(**payload)
+        )
+        await session.commit()
     return {"ok": True}
 
 
@@ -826,15 +822,9 @@ async def put_settings(payload: SettingsIn):
 
 # ── Ingestion Triggers ───────────────────────────────────────────────────
 
-def _require_public_url(url: str, field: str) -> None:
-    if not is_public_url(url):
-        raise HTTPException(400, f"{field} is not a public URL")
-
-
 @app.post("/api/ingest/rss")
 async def trigger_rss_ingest(feed_url: str, source_id: str | None = None):
     """Trigger RSS feed ingestion."""
-    _require_public_url(feed_url, "feed_url")
     count = await ingest_rss_feed(feed_url, source_id)
     return {"status": "ok", "items_ingested": count}
 
@@ -842,10 +832,6 @@ async def trigger_rss_ingest(feed_url: str, source_id: str | None = None):
 @app.post("/api/ingest/gdelt")
 async def trigger_gdelt_ingest(query: str = "", max_articles: int = 50):
     """Trigger GDELT data ingestion."""
-    _require_public_url(GDELT_API, "GDELT target")
-    parsed = urlparse(query)
-    if parsed.scheme in ("http", "https") and parsed.hostname:
-        _require_public_url(query, "query")
     count = await ingest_gdelt(query, max_articles)
     return {"status": "ok", "articles_ingested": count}
 
@@ -871,16 +857,24 @@ async def trigger_social_ingest(query: str = "", max_items: int = 50):
     return {"status": "ok", "signals_ingested": count}
 
 
+@app.post("/api/ingest/masscan")
+async def trigger_masscan(background_tasks: BackgroundTasks):
+    """Queue one masscan pass at ≤200 pps. Does not block the request on the scan."""
+    from bg_jobs import MASSCAN_PPS_CAP, schedule_masscan_pass
+
+    async def _kick() -> None:
+        schedule_masscan_pass()
+
+    background_tasks.add_task(_kick)
+    return JSONResponse(
+        {"status": "accepted", "rate_pps": MASSCAN_PPS_CAP},
+        status_code=202,
+    )
+
+
 @app.websocket("/ws/live")
 async def live_ws(ws: WebSocket):
-    """Viewport-filtered AIS/ADS-B fan-out.
-
-    Client JSON:
-      {"type":"viewport","bbox":"minlon,minlat,maxlon,maxlat"}
-      {"type":"watch_geofences","ids":["<uuid>", ...]}  — empty list = none
-    geofence_alert delivers if the point is in-viewport OR geofence_id is watched.
-    AIS/ADS-B/fire_aircraft stay viewport-only.
-    """
+    """Viewport-filtered AIS/ADS-B fan-out. Client sends {type:viewport,bbox}."""
     from ws_manager import manager
 
     client_id = str(id(ws))
@@ -906,10 +900,6 @@ async def live_ws(ws: WebSocket):
                     manager.set_viewport(client_id, parse_bbox(str(data["bbox"])))
                 except ValueError:
                     continue
-            elif data.get("type") == "watch_geofences":
-                ids = data.get("ids") or []
-                if isinstance(ids, list):
-                    manager.set_watched_geofences(client_id, [str(x) for x in ids])
     except WebSocketDisconnect:
         pass
     finally:
@@ -1061,7 +1051,7 @@ async def list_cameras(
         True,
         description="Only cameras with a verified HTTP/MJPEG snapshot_url "
                     "(the ones that actually preview). Set false to include "
-                    "rows without a snapshot_url.",
+                    "unverified masscan port-554 hits.",
     ),
     limit: int = Query(500, ge=1, le=5000),
 ):
@@ -1138,7 +1128,7 @@ async def get_camera(camera_id: UUID):
 async def camera_snapshot(camera_id: UUID):
     """Still image for one camera.
 
-    HTTP cameras go through the TTL cache. RTSP finds have no HTTP
+    HTTP cameras go through the TTL cache. masscan/RTSP finds have no HTTP
     snapshot_url — we probe common still-image paths and, failing that, grab
     one JPEG frame from RTSP via ffmpeg. No credentials are tried.
     """
@@ -1768,68 +1758,33 @@ async def api_update_geofence(gid: str, payload: GeofenceUpdate):
 @app.delete("/api/geofences/{gid}", status_code=204)
 async def api_delete_geofence(gid: str):
     from geofence import delete_geofence
-    ok = await delete_geofence(gid)
-    if not ok:
-        raise HTTPException(404, "geofence not found")
+    await delete_geofence(gid)
     return None
 
 
-@app.get("/api/geofences/{gid}/at")
-async def api_geofence_at(
-    gid: str,
-    timestamp: str = Query(..., description="ISO-8601 instant for the 1-minute DVR bucket"),
-):
-    """Aircraft/vessels/fires inside this fence at time T. Never writes."""
-    from geofence import snapshot_at
-    from tracks import parse_timestamp
-
-    try:
-        ts = parse_timestamp(timestamp)
-    except ValueError as exc:
-        raise HTTPException(422, str(exc)) from exc
-    if ts is None:
-        raise HTTPException(422, "timestamp required")
-    try:
-        body = await snapshot_at(gid, ts)
-    except Exception:
-        body = {
-            "geofence_id": gid,
-            "timestamp": ts.isoformat(),
-            "aircraft": [],
-            "vessels": [],
-            "fires": [],
-        }
-    if body is None:
-        raise HTTPException(404, "geofence not found")
-    return body
-
-
 @app.get("/api/geofence-alerts")
-async def api_geofence_alerts(
-    geofence_id: UUID | None = Query(None),
-    since: str | None = Query(None, description="ISO-8601 inclusive lower bound"),
-    until: str | None = Query(None, description="ISO-8601 inclusive upper bound"),
-    source_kind: str | None = Query(None, description="firms|ais|adsb"),
-    limit: int = Query(100, ge=1, le=500),
-):
-    """Hit log for drawn fences. Not /api/alerts (entity/keyword)."""
-    from geofence import list_alerts
-    from tracks import parse_timestamp
-
-    if source_kind is not None and source_kind not in ("firms", "ais", "adsb"):
-        raise HTTPException(422, "source_kind must be one of: firms, ais, adsb")
+async def api_geofence_alerts(limit: int = Query(100, ge=1, le=500)):
+    from sqlalchemy import text as sql_text
     try:
-        since_ts = parse_timestamp(since) if since else None
-        until_ts = parse_timestamp(until) if until else None
-    except ValueError as exc:
-        raise HTTPException(422, str(exc)) from exc
-    return await list_alerts(
-        geofence_id=str(geofence_id) if geofence_id else None,
-        since=since_ts,
-        until=until_ts,
-        source_kind=source_kind,
-        limit=limit,
-    )
+        async with async_session() as session:
+            rows = (await session.execute(sql_text(
+                """
+                SELECT id::text, geofence_id::text, source_kind, entity_id,
+                       lat, lon, payload, created_at
+                FROM geofence_alerts
+                ORDER BY created_at DESC
+                LIMIT :limit
+                """
+            ), {"limit": limit})).mappings().all()
+        out = []
+        for r in rows:
+            item = dict(r)
+            if item.get("created_at") is not None:
+                item["created_at"] = item["created_at"].isoformat()
+            out.append(item)
+        return out
+    except Exception:
+        return []
 
 
 @app.get("/api/fire-aircraft")

app/masscan_config.py

@@ -0,0 +1,65 @@
+"""Active camera-discovery configuration (masscan-based, env-driven).
+
+All knobs read from the environment with safe defaults. The scanner targets
+open TCP port 554 (RTSP — the typical IP-camera port) across a configured
+range and feeds results into the same `cameras` table as the passive scraper
+(discovery_source='masscan'), deduped by URL hash.
+
+ETHICS / SCOPE (mirrors camera_scraper.py):
+  * Detection only — a SYN port scan for OPEN hosts. No credential guessing,
+    no login attempts, no banner grabbing, and no access to camera feeds.
+  * Private / reserved ranges are excluded via MASSCAN_EXCLUDEFILE so the
+    scanner never probes RFC1918, loopback, link-local, multicast, or the
+    bogons. Fail closed if the excludefile is missing.
+
+TIMING REALITY: at the residential-safe default of 200 pps a full IPv4
+sweep (0.0.0.0/0, ~4.29B addresses) takes ~8 months. This is therefore a
+CONTINUOUS ROLLING SWEEP, not a "finish in a day" job: masscan streams
+open hosts to stdout and the runner ingests them incrementally, then
+restarts the sweep when a pass completes. New cameras are detected as they
+appear on each pass. 1k/10k pps saturated a home uplink — do not raise the
+rate unless you are on a VPS / unmetered link.
+"""
+
+from __future__ import annotations
+
+import os
+
+# Path to the masscan binary (installed on the Pi host).
+MASSCAN_BIN = os.getenv("MASSCAN_BIN", "masscan")
+
+# CIDR(s) to sweep. Default = the whole public IPv4 space.
+MASSCAN_RANGE = os.getenv("MASSCAN_RANGE", "0.0.0.0/0")
+
+# Port(s) to probe. Default 554 = RTSP, the typical IP-camera port.
+MASSCAN_PORTS = os.getenv("MASSCAN_PORTS", "554")
+
+# Packets/sec. 200 is the residential-safe default — 1k/10k pps saturated
+# a home uplink. Raise only on a VPS / unmetered link.
+MASSCAN_RATE = int(os.getenv("MASSCAN_RATE", "200"))
+
+# Retransmission count. 1 maximizes unique-host coverage at low rate; the
+# default (10) spends most of the budget re-probing the same hosts.
+MASSCAN_RETRIES = int(os.getenv("MASSCAN_RETRIES", "1"))
+
+# Seconds to keep listening for straggler responses after the last probe.
+# 0 avoids a 10s tail per pass; tiny loss of the very last hosts is fine
+# since the sweep repeats.
+MASSCAN_WAIT = int(os.getenv("MASSCAN_WAIT", "0"))
+
+# Excludefile path on the Pi host. Must contain RFC1918/loopback/link-local/
+# multicast/bogons so the scanner never probes private ranges. Fail closed if
+# the file is absent (the runner refuses to start rather than scan wide).
+MASSCAN_EXCLUDEFILE = os.getenv(
+    "MASSCAN_EXCLUDEFILE", "/etc/osint-dashboard/masscan-excludes.txt"
+)
+
+# Ingest batch size — flush this many newly-seen hosts to the DB per round.
+MASSCAN_FLUSH_EVERY = int(os.getenv("MASSCAN_FLUSH_EVERY", "250"))
+
+# NATS subject newly-found cameras are published on (same feed as the
+# passive scraper so the shared ingester persists them).
+MASSCAN_NATS_SUBJECT = os.getenv("MASSCAN_NATS_SUBJECT", "events.camera")
+
+# discovery_source tag written into the cameras table.
+MASSCAN_DISCOVERY_SOURCE = os.getenv("MASSCAN_DISCOVERY_SOURCE", "masscan")

app/masscan_scanner.py

@@ -0,0 +1,226 @@
+"""masscan result parsing + ingestion for the OSINT dashboard.
+
+Turns a stream of masscan JSON-lines (open port 554 hosts) into rows in the
+`cameras` table with discovery_source='masscan', deduped by URL hash against
+whatever the passive scraper already found. Newly discovered hosts are also
+published to NATS (`events.camera`) so the shared ingester pipeline persists
+them exactly like scraper finds.
+
+Scope: detection of OPEN hosts only. No credentials, no banners, no feed
+access. Private/reserved ranges never enter masscan (see excludefile).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from datetime import datetime, timezone
+
+from camera_models import cameras
+from camera_scraper import url_hash, geolocate_ips
+from database import async_session
+
+from masscan_config import (
+    MASSCAN_NATS_SUBJECT, MASSCAN_DISCOVERY_SOURCE,
+)
+
+logger = logging.getLogger("osint.masscan_scanner")
+
+
+# ── URL building ──────────────────────────────────────────────────────────
+
+def build_rtsp_url(ip: str) -> str:
+    """Canonical URL for an open-RTSP host. Used as the dedupe key."""
+    return f"rtsp://{ip}/"
+
+
+# ── masscan JSON parsing ──────────────────────────────────────────────────
+# masscan --output-format=json --output-file=- emits line-delimited JSON on a
+# pipe (a bare object per open host), not the array form used for seekable
+# files. We parse per-line and tolerate an accidental leading '['.
+
+def parse_masscan_line(line: str) -> list[dict]:
+    """Parse one masscan stdout line into a list of host records.
+
+    A line may contain one JSON object or, defensively, be wrapped in an
+    array. Returns [] on anything unparseable (harmless — the sweep repeats).
+    """
+    s = line.strip()
+    if not s:
+        return []
+    s = s.lstrip("[").rstrip("]").strip()
+    if not s:
+        return []
+    # Multiple records may share a line separated by '},{'.
+    if s.endswith(","):
+        s = s[:-1].rstrip()
+    out: list[dict] = []
+    for cand in _split_records(s):
+        try:
+            obj = json.loads(cand)
+        except (json.JSONDecodeError, ValueError):
+            continue
+        if isinstance(obj, dict) and obj.get("ip"):
+            out.append(obj)
+    return out
+
+
+def _split_records(s: str) -> list[str]:
+    """Split a buffer into individual JSON object strings, honoring nesting."""
+    records, depth, start = [], 0, 0
+    for i, ch in enumerate(s):
+        if ch == "{":
+            if depth == 0:
+                start = i
+            depth += 1
+        elif ch == "}":
+            depth -= 1
+            if depth == 0:
+                records.append(s[start:i + 1])
+    return records
+
+
+def extract_open_ips(records: list[dict], port: int) -> list[str]:
+    """Return the list of IPs from records that have `port` open."""
+    ips: list[str] = []
+    for rec in records:
+        for p in rec.get("ports", []):
+            if p.get("port") == port and p.get("status") == "open":
+                ips.append(rec["ip"])
+                break
+    return ips
+
+
+# ── Persistence ───────────────────────────────────────────────────────────
+
+async def ingest_open_hosts(ips: list[str]) -> tuple[int, list[str]]:
+    """Insert-or-refresh camera rows for open RTSP hosts that have a public feed.
+
+    A host only lands in the table (and therefore on the map) if an
+    unauthenticated HTTP still or MJPEG URL responds. Port-554-only hosts
+    are skipped. Returns (newly_inserted, hosts_with_working_feed).
+    """
+    if not ips:
+        return 0, []
+    from camera_preview import probe_public_feed
+
+    unique = list(dict.fromkeys(ips))
+    sem = asyncio.Semaphore(20)
+
+    async def _probe(ip: str) -> tuple[str, str | None]:
+        async with sem:
+            return ip, await probe_public_feed(ip)
+
+    probed = await asyncio.gather(*(_probe(ip) for ip in unique))
+    live = [(ip, feed) for ip, feed in probed if feed]
+    if not live:
+        logger.info("masscan ingest: 0 working feeds of %d open-554 hosts",
+                    len(unique))
+        return 0, []
+
+    now = datetime.now(timezone.utc)
+    coords = await geolocate_ips([ip for ip, _ in live])
+    new = 0
+    async with async_session() as session:
+        for ip, feed in live:
+            url = build_rtsp_url(ip)
+            h = url_hash(url)
+            lat, lon = coords.get(ip, (None, None))
+            existing = (await session.execute(
+                cameras.select().where(cameras.c.url_hash == h)
+            )).one_or_none()
+            if existing is None:
+                await session.execute(cameras.insert().values(
+                    url_hash=h,
+                    source_url=url,
+                    snapshot_url=feed,
+                    discovery_source=MASSCAN_DISCOVERY_SOURCE,
+                    location_lat=lat,
+                    location_lon=lon,
+                    location_name=f"{ip} (IP-geo)" if lat is not None else None,
+                    vendor=None,
+                    device_type="rtsp",
+                    first_seen=now,
+                    last_seen=now,
+                    raw={"discovered_via": "masscan", "port": 554,
+                         "public_feed": feed},
+                ))
+                new += 1
+            else:
+                await session.execute(cameras.update().where(
+                    cameras.c.url_hash == h
+                ).values(
+                    last_seen=now,
+                    snapshot_url=feed,
+                    location_lat=lat,
+                    location_lon=lon,
+                    location_name=f"{ip} (IP-geo)" if lat is not None else None,
+                ))
+        await session.commit()
+    logger.info("masscan ingest: %d new working feeds (%d probed, %d open-554)",
+                new, len(live), len(unique))
+    return new, [ip for ip, _ in live]
+
+
+# ── NATS publish ──────────────────────────────────────────────────────────
+
+async def publish_new_hosts(ips: list[str]) -> int:
+    """Publish newly-found open hosts to NATS for the shared ingester.
+
+    Returns the number of messages published (0 if NATS is down).
+    """
+    import json as _json
+    import nats
+    from config import NATS_URL
+
+    if not ips:
+        return 0
+    try:
+        nc = await nats.connect(NATS_URL)
+    except Exception:  # noqa: BLE001
+        logger.warning("NATS unavailable — skipping publish pass")
+        return 0
+    published = 0
+    try:
+        js = nc.jetstream()
+        for ip in dict.fromkeys(ips):
+            url = build_rtsp_url(ip)
+            msg = {
+                "source_type": "camera",
+                "title": f"Open RTSP camera ({ip})",
+                "url": url,
+                "location_lat": None,
+                "location_lon": None,
+                "location_name": None,
+                "tags": ["osint", "camera", MASSCAN_DISCOVERY_SOURCE],
+                "raw": {
+                    "url_hash": url_hash(url),
+                    "source_url": url,
+                    "snapshot_url": None,
+                    "vendor": None,
+                    "device_type": "rtsp",
+                    "discovered_via": "masscan",
+                    "port": 554,
+                },
+                "source_timestamp": datetime.now(timezone.utc).isoformat(),
+            }
+            await js.publish(MASSCAN_NATS_SUBJECT, _json.dumps(msg).encode())
+            published += 1
+    finally:
+        await nc.close()
+    logger.info("published %d masscan finds to %s", published, MASSCAN_NATS_SUBJECT)
+    return published
+
+
+# ── Batch drain helper used by the runner ─────────────────────────────────
+
+async def flush(seen: set[str], new_accum: int) -> tuple[int, int]:
+    """Ingest + publish the accumulated host set; return (new, published)."""
+    if not seen:
+        return 0, 0
+    ips = list(seen)
+    new, live = await ingest_open_hosts(ips)
+    published = await publish_new_hosts(live)
+    seen.clear()
+    return new, published

app/run_masscan_service.py

@@ -0,0 +1,149 @@
+"""Continuous masscan rolling-sweep service for the OSINT dashboard.
+
+Runs masscan against the configured range for open port 554 (RTSP), streams
+the JSON-lines output, and ingests open hosts into the `cameras` table (new
+finds only) plus publishes them to NATS — exactly like the passive scraper.
+
+Because a full IPv4 sweep at a conservative rate takes days, this runs
+masscan CONTINUOUSLY: each pass streams results in as they're found, and when
+a pass completes the sweep restarts from the top. New cameras are picked up
+on every pass.
+
+Ethics: detection-only (open-port SYN scan). Private/reserved ranges are
+excluded and the service REFUSES to start if the excludefile is missing, so
+we never probe private space by accident.
+
+Run once (for a manual/test pass):  python app/run_masscan_service.py --once
+Run forever (systemd):            python app/run_masscan_service.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import sys
+from pathlib import Path
+
+sys_path = str(Path(__file__).parent)
+sys.path.insert(0, sys_path)
+
+import masscan_config as cfg  # noqa: E402
+from database import init_extensions  # noqa: E402
+from masscan_scanner import (  # noqa: E402
+    parse_masscan_line, extract_open_ips, flush,
+)
+
+logging.basicConfig(level=logging.INFO,
+                    format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+logger = logging.getLogger("osint.masscan_service")
+
+ONCE = "--once" in sys.argv[1:]
+
+
+def _verify_excludefile() -> None:
+    """Fail closed: refuse to sweep the wide range without an excludefile."""
+    if not cfg.MASSCAN_EXCLUDEFILE:
+        raise SystemExit("MASSCAN_EXCLUDEFILE is empty — refusing to run")
+    if not Path(cfg.MASSCAN_EXCLUDEFILE).is_file():
+        raise SystemExit(
+            f"excludefile {cfg.MASSCAN_EXCLUDEFILE!r} missing — refusing to "
+            f"run (would risk probing private ranges). Install the excludefile "
+            f"first (see deploy/masscan-excludes.txt)."
+        )
+
+
+def build_command() -> list[str]:
+    cmd = [
+        cfg.MASSCAN_BIN,
+        cfg.MASSCAN_RANGE,
+        f"-p{cfg.MASSCAN_PORTS}",
+        f"--rate={cfg.MASSCAN_RATE}",
+        f"--retries={cfg.MASSCAN_RETRIES}",
+        f"--wait={cfg.MASSCAN_WAIT}",
+        "--output-format=json",
+        "--output-file=-",
+    ]
+    if cfg.MASSCAN_EXCLUDEFILE:
+        cmd.append(f"--excludefile={cfg.MASSCAN_EXCLUDEFILE}")
+    return cmd
+
+
+async def _drain_stderr(stream: asyncio.StreamReader) -> None:
+    """Consume masscan's progress chatter so its stderr pipe never fills."""
+    while True:
+        line = await stream.readline()
+        if not line:
+            break
+        text = line.decode(errors="ignore").strip()
+        if text and not text.startswith("rate:"):
+            logger.debug("masscan: %s", text)
+
+
+async def run_pass() -> tuple[int, int]:
+    """Run one full sweep pass, ingesting incrementally.
+
+    Returns (new_hosts, total_hosts_seen) for the whole pass.
+    """
+    cmd = build_command()
+    logger.info("starting masscan pass: %s", " ".join(cmd))
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    if proc.stderr is not None:
+        asyncio.ensure_future(_drain_stderr(proc.stderr))
+
+    seen: set[str] = set()
+    total_seen = 0
+    total_new = 0
+    try:
+        while True:
+            raw = await proc.stdout.readline()
+            if not raw:
+                break
+            records = parse_masscan_line(raw.decode(errors="ignore"))
+            for ip in extract_open_ips(records, 554):
+                if ip in seen:
+                    continue
+                seen.add(ip)
+                if len(seen) >= cfg.MASSCAN_FLUSH_EVERY:
+                    new, _published = await flush(seen, total_new)
+                    total_new += new
+                    total_seen += new
+        # Drain the final partial batch.
+        if seen:
+            new, _published = await flush(seen, total_new)
+            total_new += new
+        rc = await proc.wait()
+    except asyncio.CancelledError:
+        proc.kill()
+        raise
+    logger.info("masscan pass finished (rc=%s): %d new hosts ingested",
+                rc, total_new)
+    return total_new, total_seen
+
+
+async def main() -> None:
+    _verify_excludefile()
+    await init_extensions()
+    logger.info(
+        "masscan service starting: range=%s ports=%s rate=%s pps (full sweep "
+        "~%.0fh at this rate)",
+        cfg.MASSCAN_RANGE, cfg.MASSCAN_PORTS, cfg.MASSCAN_RATE,
+        4.29e9 / cfg.MASSCAN_RATE / 3600,
+    )
+    while True:
+        try:
+            await run_pass()
+        except Exception:  # noqa: BLE001
+            logger.exception("masscan pass error")
+        if ONCE:
+            return
+        # Small gap between passes so the restart is visible in logs.
+        await asyncio.sleep(5)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

app/schemas.py

@@ -7,7 +7,7 @@ from enum import Enum
 from typing import Literal, Optional
 from uuid import UUID
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, Field, field_validator
 
 
 # ─── Enums ───────────────────────────────────────────────────────────────
@@ -63,17 +63,6 @@ class FeedSourceCreate(BaseModel):
     config: Optional[dict] = None
 
 
-class FeedSourceUpdate(BaseModel):
-    """PATCH /api/sources/{id} — only these keys may be set."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    name: Optional[str] = None
-    url: Optional[str] = None
-    config: Optional[dict] = None
-    enabled: Optional[bool] = None
-
-
 class FeedSourceOut(BaseModel):
     id: UUID
     name: str

app/ws_manager.py

@@ -8,18 +8,10 @@ from __future__ import annotations
 
 import asyncio
 from typing import Any
-from uuid import UUID
 
 BBox = tuple[float, float, float, float]  # minlon, minlat, maxlon, maxlat
 
 
-def _uuid_str(value: object) -> str | None:
-    try:
-        return str(UUID(str(value)))
-    except (ValueError, TypeError, AttributeError):
-        return None
-
-
 def point_in_bbox(lon: float, lat: float, bbox: BBox | None) -> bool:
     """True if (lon, lat) sits inside an axis-aligned viewport."""
     if bbox is None:
@@ -34,7 +26,6 @@ class ConnectionManager:
     def __init__(self) -> None:
         self._queues: dict[str, asyncio.Queue] = {}
         self._viewports: dict[str, BBox] = {}
-        self._watched: dict[str, set[str]] = {}
 
     def register(self, client_id: str, maxsize: int = 256) -> asyncio.Queue:
         q: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
@@ -44,21 +35,6 @@ class ConnectionManager:
     def unregister(self, client_id: str) -> None:
         self._queues.pop(client_id, None)
         self._viewports.pop(client_id, None)
-        self._watched.pop(client_id, None)
-
-    def set_watched_geofences(self, client_id: str, ids: list[str]) -> None:
-        """Watch these fence UUIDs so geofence_alert delivers off-viewport.
-
-        Invalid UUIDs are ignored. Empty list = watch none (viewport-only).
-        """
-        if client_id not in self._queues:
-            return
-        watched: set[str] = set()
-        for raw in ids:
-            uid = _uuid_str(raw)
-            if uid is not None:
-                watched.add(uid)
-        self._watched[client_id] = watched
 
     def set_viewport(self, client_id: str, bbox: BBox) -> None:
         if client_id in self._queues:
@@ -83,21 +59,13 @@ class ConnectionManager:
     ) -> int:
         """Enqueue `{type, payload}` for clients whose viewport contains the point.
 
-        kind=geofence_alert also delivers when payload.geofence_id is in the
-        client's watch set (even if the point is off-viewport). Other kinds
-        stay viewport-only. Drops the oldest queued message if a client's
-        buffer is full. Returns the number of clients that got a copy.
+        Drops the oldest queued message if a client's buffer is full so a slow
+        tab cannot stall ingest. Returns the number of clients that got a copy.
         """
         msg = {"type": kind, "payload": payload}
         sent = 0
-        gid = _uuid_str(payload.get("geofence_id")) if kind == "geofence_alert" else None
         for client_id, queue in list(self._queues.items()):
-            in_view = point_in_bbox(lon, lat, self._viewports.get(client_id))
-            if kind == "geofence_alert":
-                watching = gid is not None and gid in self._watched.get(client_id, set())
-                if not in_view and not watching:
-                    continue
-            elif not in_view:
+            if not point_in_bbox(lon, lat, self._viewports.get(client_id)):
                 continue
             if queue.full():
                 try:

deploy/README.md

@@ -0,0 +1,31 @@
+# systemd unit template — copy to /etc/systemd/system/osint-masscan.service
+#
+# The masscan service is a CONTINUOUS rolling sweep (a full IPv4 pass at a
+# conservative rate takes ~5 days), so it runs as a long-lived service, NOT a
+# daily timer. The [Install] WantedBy means it starts at boot and Restart=always
+# keeps it up. Install steps (run once on the Pi, as root):
+#
+#   apt install -y masscan      # or: apt-get install masscan
+#   mkdir -p /etc/osint-dashboard /opt/siriusdevops
+#   cp deploy/masscan-excludes.txt /etc/osint-dashboard/masscan-excludes.txt
+#
+#   # Optional tuning (override env in this file; the DB_* values in the unit
+#   # already point at the host-published Postgres on 127.0.0.1:5432):
+#   cat > /etc/osint-dashboard/masscan.env <<'EOF'
+#   MASSCAN_RANGE=0.0.0.0/0
+#   MASSCAN_PORTS=554
+#   MASSCAN_RATE=1000
+#   EOF
+#
+#   # Venv for the scanner (host-level, not the compose image):
+#   cd /opt/siriusdevops/osint-dashboard
+#   python3 -m venv .venv-masscan
+#   .venv-masscan/bin/pip install -r app/requirements.txt
+#
+#   install -m 644 deploy/osint-masscan.service /etc/systemd/system/
+#   systemctl daemon-reload
+#   systemctl enable --now osint-masscan
+#
+# Watch:  journalctl -u osint-masscan -f
+# DB:     writes into the same Postgres the compose stack uses (127.0.0.1:5432)
+#         so findings appear on the dashboard camera map automatically.

deploy/masscan-excludes.txt

@@ -0,0 +1,33 @@
+# masscan excludefile — never probe these ranges.
+# RFC1918 private + loopback + link-local + multicast + documentation/bogons.
+# The service refuses to start if this file is missing (fail closed).
+
+# Loopback
+127.0.0.0/8
+
+# RFC1918 private
+10.0.0.0/8
+172.16.0.0/12
+192.168.0.0/16
+
+# Link-local
+169.254.0.0/16
+
+# CGNAT (RFC 6598)
+100.64.0.0/10
+
+# Multicast + reserved
+224.0.0.0/4
+240.0.0.0/4
+
+# Documentation / benchmark / example ranges (never real hosts)
+0.0.0.0/8
+192.0.2.0/24
+198.51.100.0/24
+203.0.113.0/24
+192.0.0.0/24
+198.18.0.0/15
+255.255.255.255/32
+
+# Carrier NAT / TEST-NET leftovers
+233.252.0.0/24

deploy/osint-masscan.service

@@ -0,0 +1,29 @@
+[Unit]
+Description=OSINT dashboard — masscan rolling sweep (open RTSP port 554)
+Documentation=https://forgejo.siriusdevops.com/sirius/osint-dashboard
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+# masscan needs raw sockets (CAP_NET_RAW) — run as root on the Pi host.
+User=root
+WorkingDirectory=/opt/siriusdevops/osint-dashboard
+EnvironmentFile=-/etc/osint-dashboard/masscan.env
+# Point at the compose-published Postgres on the HOST (127.0.0.1:5432), not the
+# docker service name 'postgres' which doesn't resolve outside the compose net.
+Environment=DB_HOST=127.0.0.1
+Environment=DB_PORT=5432
+Environment=DB_USER=osint
+Environment=DB_PASSWORD=osint
+Environment=DB_NAME=osint_data
+Environment=MASSCAN_EXCLUDEFILE=/etc/osint-dashboard/masscan-excludes.txt
+ExecStart=/opt/siriusdevops/osint-dashboard/.venv-masscan/bin/python app/run_masscan_service.py
+Restart=always
+RestartSec=10
+# Log the sweep to journald (read with: journalctl -u osint-masscan -f)
+StandardOutput=journal
+StandardError=journal
+
+[Install]
+WantedBy=multi-user.target

docs/free-data-streams.md

@@ -2,7 +2,7 @@
 
 Builder brief for backend + frontend. Researched 2026-08-27. Every endpoint below was either live-probed from this machine or taken from the provider’s current docs. Prefer **free, no-key, CORS-open** sources first. Keys are called out explicitly.
 
-This is **not** a camera-discovery change. Existing camera rules still apply: never emit `rtsp://` hrefs; camera pins go through `/api/cameras/{id}/snapshot`; HTTP directory cams use `/stream` MJPEG.
+This is **not** a camera-discovery / masscan change. Existing camera rules still apply: never emit `rtsp://` hrefs; masscan pins go through `/api/cameras/{id}/snapshot`; HTTP directory cams use `/stream` MJPEG.
 
 ---
 
@@ -13,7 +13,7 @@ This is **not** a camera-discovery change. Existing camera rules still apply: ne
 | NASA FIRMS VIIRS hotspots | Ingested (`app/fire_sources.py` → NATS `events.fire` → `fires` hypertable → `GET /api/fires`) | Needs free `FIRMS_MAP_KEY`. See `docs/firms.md`. |
 | NASA GIBS basemaps | Frontend tiles via `app/gibs_map.py` | No key. CORS `*`. |
 | GIBS VIIRS thermal tiles | Documented, not wired as overlay | Same GIBS stack; no key. |
-| Cameras | Scraper → `cameras` table | Defaults already include ALERTWest JPEGs + Live-Environment-Streams HLS/YouTube GeoJSON. |
+| Cameras | Scraper + masscan → `cameras` table | Defaults already include ALERTWest JPEGs + Live-Environment-Streams HLS/YouTube GeoJSON. |
 | News / RSS / GDELT / USGS quakes | Ingest | Out of scope for this brief. |
 
 **Action for existing fire ingest:** NASA will stop Suomi NPP product delivery on **2026-11-01**. Switch `FIRMS_DATASET` from `VIIRS_SNPP_NRT` to `VIIRS_NOAA20_NRT` and/or `VIIRS_NOAA21_NRT` before then.[20]
@@ -261,7 +261,7 @@ Use later if you want commuter rail / subway vehicle positions (LA Metro, MTA, e
 
 ## 6. Open video / camera feeds (official public only)
 
-Do **not** add Insecam-style random IP cams as a new source. The scraper already has a public list; this section is **agency-published** JPEG/HLS.
+Do **not** add Insecam-style random IP cams as a new source. The scraper already has a public list + masscan; this section is **agency-published** JPEG/HLS.
 
 ### 6.1 Already wired
 
@@ -304,7 +304,7 @@ Do not call the YouTube Data API unless you want search. Embedding existing stre
 
 ### 6.5 Skip
 
-- Insecam / random “public IP cam” aggregators — ToS / privacy.
+- Insecam / random “public IP cam” aggregators — ToS / privacy / already covered by masscan ethics.
 - TrafficLand, EarthCam commercial APIs.
 - SkylineWebcams — scraping, not an API.
 
@@ -523,7 +523,7 @@ Attribution bar (required): OpenSky / ADSB.lol ODbL / Amtraker / RainViewer / IE
 
 ## 12. Legal / ethics (non-negotiable)
 
-- RTSP policy unchanged (never emit `rtsp://` hrefs).
+- Masscan / RTSP policy unchanged.
 - AISStream: server-side only; do not put the key in JS.[5]
 - OpenSky: non-commercial unless licensed; cite if you publish.[2]
 - ADSB.lol: ODbL share-alike on derived databases.[4]

tests/test_bg_jobs.py

@@ -1,4 +1,4 @@
-"""ffmpeg snapshots stay off the request path (asyncio.create_task)."""
+"""masscan/ffmpeg stay off the request path (asyncio.create_task)."""
 
 from __future__ import annotations
 
@@ -7,27 +7,36 @@ import asyncio
 import bg_jobs
 
 
-def test_bg_jobs_has_no_pps_cap():
-    assert not any(name.endswith("_PPS_CAP") for name in dir(bg_jobs))
+def test_schedule_masscan_pass_returns_without_awaiting_scan(monkeypatch):
+    started = {"n": 0}
+
+    async def slow_pass():
+        started["n"] += 1
+        await asyncio.sleep(30)
+
+    monkeypatch.setattr(bg_jobs, "_run_masscan_capped", slow_pass)
+    bg_jobs._masscan_task = None
+
+    async def run():
+        launched = bg_jobs.schedule_masscan_pass()
+        assert launched is True
+        # Must not have blocked for the 30s pass.
+        assert bg_jobs._masscan_task is not None
+        assert not bg_jobs._masscan_task.done()
+        launched2 = bg_jobs.schedule_masscan_pass()
+        assert launched2 is False  # already running
+        bg_jobs._masscan_task.cancel()
+        try:
+            await bg_jobs._masscan_task
+        except (asyncio.CancelledError, Exception):
+            pass
+        bg_jobs._masscan_task = None
+
+    asyncio.run(run())
 
 
-def test_camera_preview_has_no_public_feed_probe():
-    import camera_preview
-
-    assert not hasattr(camera_preview, "probe_public_feed")
-    assert not hasattr(camera_preview, "_http_feed_url")
-
-
-def test_ingest_routes_exclude_active_discovery():
-    from main import app
-
-    ingest = [
-        getattr(r, "path", "")
-        for r in app.routes
-        if getattr(r, "path", "").startswith("/api/ingest/")
-    ]
-    assert "/api/ingest/fires" in ingest
-    assert all("scan" not in path for path in ingest)
+def test_masscan_rate_cap_is_200():
+    assert bg_jobs.MASSCAN_PPS_CAP == 200
 
 
 def test_schedule_ffmpeg_snapshot_is_a_task_not_inline(monkeypatch):

tests/test_geofence.py

@@ -51,33 +51,23 @@ def test_matching_geofences_only_active_hits():
     assert matching_geofences(-122.4, 37.7, fences) == []
 
 
-FENCE_ID = "11111111-1111-1111-1111-111111111111"
-NC_VIEW = (-80.0, 35.0, -78.0, 36.0)
-SF_VIEW = (-123.0, 37.0, -121.0, 38.0)
-
-
-def _alert_payload(gid=FENCE_ID):
-    return {
-        "geofence_id": gid,
-        "geofence_name": "NC",
-        "source_kind": "ais",
-        "entity_id": "366123456",
-        "lat": 35.5,
-        "lon": -79.0,
-    }
-
-
 def test_geofence_alert_fans_out_only_to_viewport_clients():
     mgr = ConnectionManager()
     q_nc = mgr.register("nc")
     q_sf = mgr.register("sf")
-    mgr.set_viewport("nc", NC_VIEW)
-    mgr.set_viewport("sf", SF_VIEW)
+    mgr.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0))
+    mgr.set_viewport("sf", (-123.0, 37.0, -121.0, 38.0))
 
     async def run():
-        n = await mgr.publish_point(
-            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
-        )
+        payload = {
+            "geofence_id": "a",
+            "geofence_name": "NC",
+            "source_kind": "ais",
+            "entity_id": "366123456",
+            "lat": 35.5,
+            "lon": -79.0,
+        }
+        n = await mgr.publish_point("geofence_alert", payload, lat=35.5, lon=-79.0)
         assert n == 1
         msg = q_nc.get_nowait()
         assert msg["type"] == "geofence_alert"
@@ -87,109 +77,6 @@ def test_geofence_alert_fans_out_only_to_viewport_clients():
     asyncio.run(run())
 
 
-def test_off_viewport_watch_receives_geofence_alert():
-    mgr = ConnectionManager()
-    q_sf = mgr.register("sf")
-    mgr.set_viewport("sf", SF_VIEW)
-    mgr.set_watched_geofences("sf", [FENCE_ID])
-
-    async def run():
-        n = await mgr.publish_point(
-            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
-        )
-        assert n == 1
-        msg = q_sf.get_nowait()
-        assert msg["type"] == "geofence_alert"
-        assert msg["payload"]["geofence_id"] == FENCE_ID
-
-    asyncio.run(run())
-
-
-def test_off_viewport_without_watch_does_not_receive_geofence_alert():
-    mgr = ConnectionManager()
-    q_sf = mgr.register("sf")
-    mgr.set_viewport("sf", SF_VIEW)
-
-    async def run():
-        n = await mgr.publish_point(
-            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
-        )
-        assert n == 0
-        assert q_sf.empty()
-
-    asyncio.run(run())
-
-
-def test_on_viewport_receives_geofence_alert_without_watch():
-    mgr = ConnectionManager()
-    q_nc = mgr.register("nc")
-    mgr.set_viewport("nc", NC_VIEW)
-
-    async def run():
-        n = await mgr.publish_point(
-            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
-        )
-        assert n == 1
-        assert q_nc.get_nowait()["type"] == "geofence_alert"
-
-    asyncio.run(run())
-
-
-def test_ais_stays_viewport_only_even_when_watching():
-    mgr = ConnectionManager()
-    q_sf = mgr.register("sf")
-    mgr.set_viewport("sf", SF_VIEW)
-    mgr.set_watched_geofences("sf", [FENCE_ID])
-
-    async def run():
-        n = await mgr.publish_point("ais", {"id": "366123456"}, lat=35.5, lon=-79.0)
-        assert n == 0
-        assert q_sf.empty()
-
-    asyncio.run(run())
-
-
-def test_invalid_watch_uuids_ignored_empty_list_clears():
-    mgr = ConnectionManager()
-    q = mgr.register("sf")
-    mgr.set_viewport("sf", SF_VIEW)
-    mgr.set_watched_geofences("sf", ["not-a-uuid", FENCE_ID, "also-bad"])
-
-    async def run():
-        n = await mgr.publish_point(
-            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
-        )
-        assert n == 1
-        q.get_nowait()
-        mgr.set_watched_geofences("sf", [])
-        n2 = await mgr.publish_point(
-            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
-        )
-        assert n2 == 0
-        assert q.empty()
-
-    asyncio.run(run())
-
-
-def test_unregister_clears_watched_geofences():
-    mgr = ConnectionManager()
-    q = mgr.register("sf")
-    mgr.set_viewport("sf", SF_VIEW)
-    mgr.set_watched_geofences("sf", [FENCE_ID])
-    mgr.unregister("sf")
-    q2 = mgr.register("sf")
-    mgr.set_viewport("sf", SF_VIEW)
-
-    async def run():
-        n = await mgr.publish_point(
-            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
-        )
-        assert n == 0
-        assert q2.empty()
-
-    asyncio.run(run())
-
-
 def test_record_and_notify_queries_postgis_when_cache_empty(monkeypatch):
     """FIRMS ingest in the ingester has an empty in-process cache — still ST_Intersects."""
     import geofence
@@ -247,117 +134,3 @@ def test_record_and_notify_queries_postgis_when_cache_empty(monkeypatch):
     inserts = [p for p in executed if isinstance(p, dict)]
     assert inserts and inserts[0]["source_kind"] == "firms"
     assert "commit" in executed
-
-
-def test_list_alerts_sql_filters(monkeypatch):
-    captured: dict = {}
-
-    class FakeResult:
-        def mappings(self):
-            return self
-
-        def all(self):
-            return []
-
-    class FakeSession:
-        async def execute(self, stmt, params=None):
-            captured["sql"] = str(stmt)
-            captured["params"] = params
-            return FakeResult()
-
-        async def __aenter__(self):
-            return self
-
-        async def __aexit__(self, *a):
-            return False
-
-    monkeypatch.setattr(geofence, "async_session", FakeSession)
-    from datetime import datetime, timezone
-
-    since = datetime(2026, 8, 28, tzinfo=timezone.utc)
-    until = datetime(2026, 8, 29, tzinfo=timezone.utc)
-
-    async def run():
-        return await geofence.list_alerts(
-            geofence_id=FENCE_ID, since=since, until=until,
-            source_kind="firms", limit=5,
-        )
-
-    assert asyncio.run(run()) == []
-    sql = captured["sql"].lower()
-    assert "geofence_id" in sql
-    assert "created_at >=" in sql
-    assert "created_at <=" in sql
-    assert "source_kind" in sql
-    assert captured["params"]["geofence_id"] == FENCE_ID
-    assert captured["params"]["source_kind"] == "firms"
-    assert captured["params"]["limit"] == 5
-
-
-def test_alembic_fence_created_index_exists():
-    from pathlib import Path
-    text = Path(__file__).resolve().parent.parent.joinpath(
-        "alembic/versions/011_geofence_alerts_fence.py",
-    ).read_text()
-    assert "ix_geofence_alerts_fence_created" in text
-    assert "010_bbox_gist" in text
-
-
-def test_snapshot_at_404_when_fence_missing(monkeypatch):
-    geofence._cache.clear()
-
-    async def boom():
-        raise RuntimeError("db down")
-
-    monkeypatch.setattr(geofence, "refresh_cache", boom)
-
-    async def run():
-        from datetime import datetime, timezone
-        return await geofence.snapshot_at(
-            FENCE_ID, datetime(2026, 8, 28, 12, 4, tzinfo=timezone.utc),
-        )
-
-    assert asyncio.run(run()) is None
-
-
-def test_snapshot_queries_st_intersects(monkeypatch):
-    geofence._cache[:] = [{
-        "id": FENCE_ID, "name": "NC", "geojson": NC_BOX, "active": True,
-    }]
-    sqls: list[str] = []
-
-    class FakeResult:
-        def mappings(self):
-            return self
-
-        def all(self):
-            return []
-
-    class FakeSession:
-        async def execute(self, stmt, params=None):
-            sqls.append(str(stmt))
-            return FakeResult()
-
-        async def __aenter__(self):
-            return self
-
-        async def __aexit__(self, *a):
-            return False
-
-    monkeypatch.setattr(geofence, "async_session", FakeSession)
-
-    async def run():
-        from datetime import datetime, timezone
-        return await geofence.snapshot_at(
-            FENCE_ID, datetime(2026, 8, 28, 12, 4, 30, tzinfo=timezone.utc),
-        )
-
-    body = asyncio.run(run())
-    assert body["aircraft"] == []
-    assert body["vessels"] == []
-    assert body["fires"] == []
-    blob = "\n".join(sqls).lower()
-    assert "st_intersects" in blob
-    assert "aircraft_tracks_1min" in blob
-    assert "vessel_tracks_1min" in blob
-    assert "from fires" in blob

tests/test_geofence_frontend.py

@@ -1,4 +1,4 @@
-"""Geofence layer panel: draw, watch, inbox, delete (HTML contract)."""
+"""Geofence layer panel: draw + delete (DELETE /api/geofences/{id})."""
 
 from __future__ import annotations
 
@@ -24,33 +24,3 @@ def test_load_geofences_renders_delete_controls():
     assert "deleteGeofence" in js
     assert "onEachFeature" in js
     assert "bindPopup" in js
-
-
-def test_finish_cancel_draw_controls():
-    assert 'id="gf-finish"' in HTML
-    assert 'id="gf-cancel"' in HTML
-    assert "function cancelGeofenceDraw" in HTML
-    assert "function onGfClose" in HTML
-
-
-def test_watch_geofences_ws_payload():
-    assert "watch_geofences" in HTML
-    assert "function sendWatchGeofences" in HTML
-
-
-def test_geofence_alert_inbox():
-    assert 'id="gf-inbox"' in HTML
-    assert "/api/geofence-alerts" in HTML
-    assert "function loadGfInbox" in HTML
-    assert "function pushGfInbox" in HTML
-
-
-def test_delete_geofence_still_present():
-    assert "function deleteGeofence" in HTML
-    assert "method: 'DELETE'" in HTML or 'method: "DELETE"' in HTML
-
-
-def test_fence_dvr_at_endpoint():
-    assert "/at?timestamp=" in HTML or "/at?timestamp=${" in HTML
-    assert "function dvrScrubFence" in HTML
-    assert "gfSelectedId" in HTML

tests/test_hud_chrome.py

@@ -1,113 +0,0 @@
-"""Quiet HUD chrome: VIIRS default, collapsed rail, no Orbitron/MKT dashes."""
-
-from __future__ import annotations
-
-from pathlib import Path
-
-ROOT = Path(__file__).resolve().parent.parent
-HTML = (ROOT / "app/static/index.html").read_text()
-
-
-def _attr(html: str, elem_id: str) -> str:
-    chunk = html.split(f'id="{elem_id}"', 1)[1].split(">", 1)[0]
-    return chunk
-
-
-def test_initmap_prefers_viirs_true_color():
-    init = HTML.split("async function initMap", 1)[1].split("function readMapPrefs", 1)[0]
-    assert "VIIRS_SNPP_CorrectedReflectance_TrueColor" in init
-    assert init.index("VIIRS_SNPP_CorrectedReflectance_TrueColor") < init.index(
-        "MODIS_Terra_CorrectedReflectance_TrueColor"
-    )
-    assert init.index("MODIS_Terra_CorrectedReflectance_TrueColor") < init.index(
-        "BlueMarble_ShadedRelief_Bathymetry"
-    )
-
-
-def test_orbitron_gone():
-    assert "Orbitron" not in HTML
-    assert "IBM Plex Sans" in HTML
-    assert "IBM Plex Mono" in HTML
-
-
-def test_lp_note_stripped_from_layer_list():
-    assert 'class="lp-note"' not in HTML
-    body = HTML.split('class="lp-body"', 1)[1].split("lp-legend", 1)[0]
-    assert "lp-note" not in body
-
-
-def test_default_overlays_basemap_and_firms_only():
-    fires = _attr(HTML, "lp-fires-on")
-    assert "checked" in fires
-    for eid in (
-        "lp-cams-on",
-        "lp-blips-on",
-        "lp-news-on",
-        "lp-radar-on",
-        "lp-alerts-on",
-        "lp-perim-on",
-        "lp-ac-on",
-        "lp-trains-on",
-        "lp-storms-on",
-    ):
-        assert "checked" not in _attr(HTML, eid), eid
-
-
-def test_geofence_markup_before_cameras():
-    assert 'id="gf-draw"' in HTML
-    assert HTML.index('id="gf-draw"') < HTML.index('id="lp-cams-on"')
-    assert HTML.index('id="lp-base-on"') < HTML.index('id="gf-draw"')
-
-
-def test_parent_geofence_hud_survives():
-    assert "watch_geofences" in HTML
-    assert "function deleteGeofence" in HTML
-    assert 'id="gf-finish"' in HTML
-    assert 'id="gf-cancel"' in HTML
-    assert 'id="gf-inbox"' in HTML
-
-
-def test_layer_rail_collapsed_on_load():
-    head = HTML.split('class="lp-head"', 1)[1].split("</div>", 1)[0]
-    assert 'aria-expanded="false"' in head
-    assert 'id="layer-panel" class="collapsed"' in HTML
-
-
-def test_market_ticker_hidden_no_poll():
-    mkt = HTML.split('class="ticker market"', 1)[1].split(">", 1)[0]
-    assert "hidden" in mkt
-    assert "setInterval(probeMarket" not in HTML
-    assert "setInterval(loadMarket" not in HTML
-    init = HTML.split("function initMarketTicker", 1)[1].split("function ", 1)[0]
-    assert "/api/market" in init or "404-poll" in init
-    assert "setInterval" not in init
-
-
-def test_news_ticker_fills_news_only_dock():
-    css = HTML.split("</style>", 1)[0]
-    compact = css.replace(" ", "").replace("\n", "")
-    assert ".dock.news-only{height:32px;}" in compact
-    assert ".dock.news-only.ticker{height:100%;}" in compact
-    assert ".ticker{display:flex;align-items:stretch;height:50%;" in compact
-
-
-def test_news_pins_are_circle_markers():
-    js = HTML.split("async function loadNewsPins", 1)[1].split("function refreshLiveOverlays", 1)[0]
-    assert "L.circleMarker" in js
-    assert "fillOpacity: 0.7" in js or "fillOpacity:0.7" in js
-    assert "rotate(45deg)" not in js
-    assert "L.divIcon" not in js
-
-
-def test_chokepoint_buttons_not_in_toolbar_flow():
-    assert 'id="chokepoint-select"' in HTML
-    css = HTML.split("</style>", 1)[0]
-    assert ".chokepoint-btns { display: none; }" in css or ".chokepoint-btns{display:none" in css.replace(
-        " ", ""
-    )
-
-
-def test_brand_is_osint_slash():
-    assert "GLOBAL SITUATIONAL AWARENESS TERMINAL" not in HTML
-    assert "OSINT" in HTML
-    assert 'class="accent">//</span>' in HTML

tests/test_hud_osiris.py

@@ -60,7 +60,7 @@ def test_camera_thumbs_gated_at_zoom_12():
     assert "camThumbsAllowed" in thumb or "CAM_THUMB_MIN_ZOOM" in thumb
     assert "zoom in for preview" in HTML or "zoom for preview" in HTML
     assert "preview unavailable" in HTML
-    # RTSP still proxy through snapshot; never emit rtsp hrefs.
+    # Masscan / RTSP still proxy through snapshot; never emit rtsp hrefs.
     src = _fn("camSourceLink", "youtubeId")
     assert "rtsp://" in src
     assert "href=" not in src.split("rtsp://")[1].split("return")[0] or "Never emit" in src

tests/test_ingest_ssrf_and_sources.py

@@ -1,67 +0,0 @@
-"""SSRF guard on ingest triggers + PATCH /api/sources allowlist."""
-
-from __future__ import annotations
-
-import asyncio
-
-import httpx
-import pytest
-from pydantic import ValidationError
-
-from main import app
-
-BASE = "http://test"
-LINK_LOCAL_META = "http://169.254.169.254/latest/meta-data/"
-LOOPBACK = "http://127.0.0.1/secret"
-
-
-async def _req(method: str, path: str, **kw) -> httpx.Response:
-    transport = httpx.ASGITransport(app=app)
-    async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
-        return await client.request(method, path, **kw)
-
-
-def test_rss_ingest_rejects_link_local_metadata_url(monkeypatch):
-    called = {"n": 0}
-
-    async def _boom(*_a, **_k):
-        called["n"] += 1
-        raise AssertionError("ingest_rss_feed must not run for a private URL")
-
-    monkeypatch.setattr("main.ingest_rss_feed", _boom)
-    resp = asyncio.run(_req("POST", "/api/ingest/rss", params={"feed_url": LINK_LOCAL_META}))
-    assert resp.status_code == 400
-    assert called["n"] == 0
-
-
-def test_gdelt_ingest_rejects_private_query_url(monkeypatch):
-    called = {"n": 0}
-
-    async def _boom(*_a, **_k):
-        called["n"] += 1
-        raise AssertionError("ingest_gdelt must not run for a private URL query")
-
-    monkeypatch.setattr("main.ingest_gdelt", _boom)
-    resp = asyncio.run(_req("POST", "/api/ingest/gdelt", params={"query": LOOPBACK}))
-    assert resp.status_code == 400
-    assert called["n"] == 0
-
-
-def test_update_source_rejects_unknown_fields():
-    sid = "00000000-0000-0000-0000-000000000001"
-    resp = asyncio.run(_req("PATCH", f"/api/sources/{sid}", json={"enabled": True, "source_type": "rss"}))
-    assert resp.status_code == 422
-
-
-def test_feed_source_update_allowlist_only():
-    from schemas import FeedSourceUpdate
-
-    payload = FeedSourceUpdate(name="n", url="https://example.com/rss", config={"k": 1}, enabled=False)
-    assert payload.model_dump(exclude_unset=True) == {
-        "name": "n",
-        "url": "https://example.com/rss",
-        "config": {"k": 1},
-        "enabled": False,
-    }
-    with pytest.raises(ValidationError):
-        FeedSourceUpdate.model_validate({"enabled": True, "id": "00000000-0000-0000-0000-000000000001"})

tests/test_phase2_api.py

@@ -17,9 +17,6 @@ async def _req(method: str, path: str, **kw) -> httpx.Response:
         return await client.request(method, path, **kw)
 
 
-FENCE_ID = "11111111-1111-1111-1111-111111111111"
-
-
 def test_geofence_post_rejects_point():
     resp = asyncio.run(_req(
         "POST", "/api/geofences",
@@ -28,111 +25,6 @@ def test_geofence_post_rejects_point():
     assert resp.status_code == 422
 
 
-def test_delete_geofence_404_when_missing(monkeypatch):
-    async def missing(_gid: str) -> bool:
-        return False
-
-    monkeypatch.setattr("geofence.delete_geofence", missing)
-    resp = asyncio.run(_req("DELETE", f"/api/geofences/{FENCE_ID}"))
-    assert resp.status_code == 404
-
-
-def test_geofence_alerts_passes_filters(monkeypatch):
-    seen = {}
-
-    async def fake_list(**kwargs):
-        seen.update(kwargs)
-        return [{"id": "a", "geofence_id": FENCE_ID, "source_kind": "ais"}]
-
-    monkeypatch.setattr("geofence.list_alerts", fake_list)
-    resp = asyncio.run(_req(
-        "GET", "/api/geofence-alerts",
-        params={
-            "geofence_id": FENCE_ID,
-            "since": "2026-08-28T00:00:00Z",
-            "until": "2026-08-29T00:00:00Z",
-            "source_kind": "ais",
-            "limit": 10,
-        },
-    ))
-    assert resp.status_code == 200
-    assert resp.json()[0]["source_kind"] == "ais"
-    assert seen["geofence_id"] == FENCE_ID
-    assert seen["source_kind"] == "ais"
-    assert seen["limit"] == 10
-    assert seen["since"] is not None
-    assert seen["until"] is not None
-
-
-def test_geofence_alerts_rejects_bad_source_kind():
-    resp = asyncio.run(_req(
-        "GET", "/api/geofence-alerts", params={"source_kind": "camera"},
-    ))
-    assert resp.status_code == 422
-
-
-def test_geofence_at_404_when_missing(monkeypatch):
-    async def no_snap(gid: str, ts):
-        return None
-
-    monkeypatch.setattr("geofence.snapshot_at", no_snap)
-    resp = asyncio.run(_req(
-        "GET", f"/api/geofences/{FENCE_ID}/at",
-        params={"timestamp": "2026-08-28T12:04:00Z"},
-    ))
-    assert resp.status_code == 404
-
-
-def test_geofence_at_empty_lists_when_db_down(monkeypatch):
-    async def empty_snap(gid: str, ts):
-        return {
-            "geofence_id": gid,
-            "timestamp": ts.isoformat(),
-            "aircraft": [],
-            "vessels": [],
-            "fires": [],
-        }
-
-    monkeypatch.setattr("geofence.snapshot_at", empty_snap)
-    resp = asyncio.run(_req(
-        "GET", f"/api/geofences/{FENCE_ID}/at",
-        params={"timestamp": "2026-08-28T12:04:00Z"},
-    ))
-    assert resp.status_code == 200
-    body = resp.json()
-    assert body["geofence_id"] == FENCE_ID
-    assert body["aircraft"] == []
-    assert body["vessels"] == []
-    assert body["fires"] == []
-    assert "timestamp" in body
-
-
-def test_geofence_at_does_not_notify(monkeypatch):
-    called = {"notify": 0}
-
-    async def empty_snap(gid: str, ts):
-        return {
-            "geofence_id": gid,
-            "timestamp": ts.isoformat(),
-            "aircraft": [],
-            "vessels": [],
-            "fires": [],
-        }
-
-    async def boom(**_kw):
-        called["notify"] += 1
-        raise AssertionError("GET /at must not record_and_notify")
-
-    monkeypatch.setattr("geofence.snapshot_at", empty_snap)
-    monkeypatch.setattr("geofence.record_and_notify", boom)
-    resp = asyncio.run(_req(
-        "GET", f"/api/geofences/{FENCE_ID}/at",
-        params={"timestamp": "2026-08-28T12:04:00Z"},
-    ))
-    assert resp.status_code == 200
-    assert called["notify"] == 0
-
-
 def test_geofences_list_does_not_collide_with_alerts():
     resp = asyncio.run(_req("GET", "/api/geofences"))
     assert resp.status_code == 200