Merge branch 'master' into feat/ssrf-ingest-source-whitelist

Reviewed-on: #45

.env.example

@@ -31,20 +31,6 @@ 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/010_bbox_gist.py

@@ -0,0 +1,29 @@
+"""GIST bbox indexes for events/fires map-pan queries.
+
+Revision ID: 010_bbox_gist
+Revises: 009_vessels
+Create Date: 2026-09-01
+"""
+
+from alembic import op
+
+revision = "010_bbox_gist"
+down_revision = "009_vessels"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_events_geom_gist ON events "
+        "USING gist (ST_SetSRID(ST_MakePoint(location_lon, location_lat), 4326))"
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_fires_geom_gist ON fires "
+        "USING gist (ST_SetSRID(ST_MakePoint(longitude, latitude), 4326))"
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS ix_fires_geom_gist")
+    op.execute("DROP INDEX IF EXISTS ix_events_geom_gist")

app/bg_jobs.py

@@ -1,50 +1,19 @@
-"""Background masscan / ffmpeg — never block a FastAPI request on a scan.
+"""Background ffmpeg — never block a FastAPI request on a frame grab.
 
-masscan is capped at 200 pps (home uplink saturates at 1k+). ffmpeg frame
-grabs are scheduled with asyncio.create_task and shared per URL.
+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. masscan finds are stored as `rtsp://IP/` with no snapshot_url, so
+stream. Some scraper sources store `rtsp://` URLs 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,16 +42,6 @@ _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")
 
 
@@ -87,55 +77,6 @@ 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/main.py

@@ -26,7 +26,7 @@ from uuid import UUID
 
 import httpx
 import structlog
-from fastapi import BackgroundTasks, FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect
+from fastapi import FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect
 from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
@@ -871,21 +871,6 @@ 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 sends {type:viewport,bbox}."""
@@ -1065,7 +1050,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 "
-                    "unverified masscan port-554 hits.",
+                    "rows without a snapshot_url.",
     ),
     limit: int = Query(500, ge=1, le=5000),
 ):
@@ -1142,7 +1127,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. masscan/RTSP finds have no HTTP
+    HTTP cameras go through the TTL cache. 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.
     """
@@ -2006,4 +1991,4 @@ app.mount("/static", CachedStaticFiles(directory=str(STATIC_DIR)), name="static"
 
 if __name__ == "__main__":
     import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+    uvicorn.run(app, host="0.0.0.0", port=8000, workers=1)  # single worker: in-memory WS/pubsub + layer caches

app/masscan_config.py

@@ -1,65 +0,0 @@
-"""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

@@ -1,226 +0,0 @@
-"""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

@@ -1,149 +0,0 @@
-"""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())

deploy/README.md

@@ -1,31 +0,0 @@
-# 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

@@ -1,33 +0,0 @@
-# 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

@@ -1,29 +0,0 @@
-[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

docker-compose.yml

@@ -170,7 +170,7 @@ services:
   # Listens on 8000 INSIDE the container (the app already owns host 8000);
   # published on host loopback 127.0.0.1:8001 only.
   titiler:
-    image: ghcr.io/developmentseed/titiler:latest
+    image: ghcr.io/developmentseed/titiler:latest@sha256:1809958d063543e3ec858259536002b2de78e9f8f09a22a8d9591bdc2b550b14
     pull_policy: missing
     container_name: osint-titiler
     platform: linux/arm64

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 / 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.
+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.
 
 ---
 
@@ -13,7 +13,7 @@ This is **not** a camera-discovery / masscan change. Existing camera rules still
 | 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 + masscan → `cameras` table | Defaults already include ALERTWest JPEGs + Live-Environment-Streams HLS/YouTube GeoJSON. |
+| Cameras | Scraper → `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 + masscan; 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; 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 / already covered by masscan ethics.
+- Insecam / random “public IP cam” aggregators — ToS / privacy.
 - 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)
 
-- Masscan / RTSP policy unchanged.
+- RTSP policy unchanged (never emit `rtsp://` hrefs).
 - 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 @@
-"""masscan/ffmpeg stay off the request path (asyncio.create_task)."""
+"""ffmpeg snapshots stay off the request path (asyncio.create_task)."""
 
 from __future__ import annotations
 
@@ -7,36 +7,27 @@ import asyncio
 import 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_bg_jobs_has_no_pps_cap():
+    assert not any(name.endswith("_PPS_CAP") for name in dir(bg_jobs))
 
 
-def test_masscan_rate_cap_is_200():
-    assert bg_jobs.MASSCAN_PPS_CAP == 200
+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_schedule_ffmpeg_snapshot_is_a_task_not_inline(monkeypatch):

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
-    # Masscan / RTSP still proxy through snapshot; never emit rtsp hrefs.
+    # 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_phase1_guardrails.py

@@ -31,3 +31,35 @@ def test_no_redis_kafka_celery():
         assert "kafka" not in blob
         assert "celery" not in blob
     assert "cachetools" in req
+
+
+def test_titiler_image_pinned_by_digest():
+    text = (ROOT / "docker-compose.yml").read_text()
+    assert (
+        "ghcr.io/developmentseed/titiler:latest@sha256:"
+        "1809958d063543e3ec858259536002b2de78e9f8f09a22a8d9591bdc2b550b14"
+        in text
+    )
+    # Unpinned :latest would drift on every pull.
+    for line in text.splitlines():
+        if "titiler" in line.lower() and "image:" in line:
+            assert "@sha256:" in line
+
+
+def test_uvicorn_single_worker_guard():
+    text = (ROOT / "app" / "main.py").read_text()
+    main_block = text.split('if __name__ == "__main__":', 1)[1]
+    assert "workers=1" in main_block
+
+
+def test_bbox_gist_migration_keeps_btree_and_adds_gist():
+    text = (ROOT / "alembic" / "versions" / "010_bbox_gist.py").read_text()
+    assert "down_revision" in text and "009_vessels" in text
+    assert "ix_events_geom_gist" in text
+    assert "ix_fires_geom_gist" in text
+    assert "ST_MakePoint(location_lon, location_lat)" in text
+    assert "ST_MakePoint(longitude, latitude)" in text
+    assert "USING gist" in text
+    models = (ROOT / "app" / "models.py").read_text()
+    assert 'Index("ix_events_location"' in models
+    assert 'Index("ix_fires_bbox"' in models