masscan: active RTSP (554) camera discovery service

Continuous whole-IPv4 rolling sweep for open TCP 554, feeding the same
cameras table as the passive scraper (discovery_source=masscan).

- masscan_config.py: env-driven knobs (range, ports, rate, retries, excludes)
- masscan_scanner.py: JSON-lines parser, rtsp://IP/ URL + url_hash dedupe,
  ip-api geolocation, insert/refresh, NATS publish for new finds
- run_masscan_service.py: long-lived rolling-sweep runner (streams results
  in, restarts on pass completion); fails closed without an excludefile
- deploy/: systemd unit + README + excludes file for the Pi host
- .env.example: masscan section

Verified end-to-end against a local Postgres: parse, insert, and dedupe
(0 new on re-ingest) all pass.

.env.example

@@ -28,6 +28,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: at a conservative 10000 pps a full 0.0.0.0/0 sweep takes ~5 days, so
+# this is a continuous rolling sweep, not a daily job.
+MASSCAN_RANGE=0.0.0.0/0
+MASSCAN_PORTS=554
+MASSCAN_RATE=10000
+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.

app/masscan_config.py

@@ -0,0 +1,64 @@
+"""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 default conservative rate of 10,000 pps a full IPv4
+sweep (0.0.0.0/0, ~4.29B addresses) takes ~119 hours (~5 days). 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.
+"""
+
+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. 10,000 = conservative, polite, residential-IP friendly
+# (~5 days for a full sweep). Raise carefully on a capable host/VPS.
+MASSCAN_RATE = int(os.getenv("MASSCAN_RATE", "10000"))
+
+# 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,207 @@
+"""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, int]:
+    """Insert-or-refresh camera rows for open RTSP hosts.
+
+    Returns (newly_inserted, total_hosts_seen_this_batch). Geolocates each
+    host via the shared ip-api batch resolver. Already-known hosts (matching
+    url_hash) have last_seen/coords refreshed and are NOT counted as new.
+    """
+    if not ips:
+        return 0, 0
+    now = datetime.now(timezone.utc)
+    coords = await geolocate_ips(list(dict.fromkeys(ips)))
+    new = 0
+    async with async_session() as session:
+        for ip in dict.fromkeys(ips):
+            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=None,          # RTSP-only; no HTTP snapshot
+                    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},
+                ))
+                new += 1
+            else:
+                await session.execute(cameras.update().where(
+                    cameras.c.url_hash == h
+                ).values(
+                    last_seen=now,
+                    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, %d refreshed", new, len(set(ips)))
+    return new, len(set(ips))
+
+
+# ── 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, _ = await ingest_open_hosts(ips)
+    published = await publish_new_hosts(ips)
+    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())

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=10000
+#   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