From 085061492eed0ac936aad173ebc7151612f78400 Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Mon, 24 Aug 2026 22:23:36 -0400 Subject: [PATCH] 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 | 14 +++ app/masscan_config.py | 64 +++++++++++ app/masscan_scanner.py | 207 +++++++++++++++++++++++++++++++++++ app/run_masscan_service.py | 149 +++++++++++++++++++++++++ deploy/README.md | 31 ++++++ deploy/masscan-excludes.txt | 33 ++++++ deploy/osint-masscan.service | 29 +++++ 7 files changed, 527 insertions(+) create mode 100644 app/masscan_config.py create mode 100644 app/masscan_scanner.py create mode 100644 app/run_masscan_service.py create mode 100644 deploy/README.md create mode 100644 deploy/masscan-excludes.txt create mode 100644 deploy/osint-masscan.service diff --git a/.env.example b/.env.example index ae8fd22..1636813 100644 --- a/.env.example +++ b/.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. diff --git a/app/masscan_config.py b/app/masscan_config.py new file mode 100644 index 0000000..81df645 --- /dev/null +++ b/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") diff --git a/app/masscan_scanner.py b/app/masscan_scanner.py new file mode 100644 index 0000000..347a714 --- /dev/null +++ b/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 diff --git a/app/run_masscan_service.py b/app/run_masscan_service.py new file mode 100644 index 0000000..2628576 --- /dev/null +++ b/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()) diff --git a/deploy/README.md b/deploy/README.md new file mode 100644 index 0000000..dbb5eae --- /dev/null +++ b/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. diff --git a/deploy/masscan-excludes.txt b/deploy/masscan-excludes.txt new file mode 100644 index 0000000..e93defb --- /dev/null +++ b/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 diff --git a/deploy/osint-masscan.service b/deploy/osint-masscan.service new file mode 100644 index 0000000..ce6b34c --- /dev/null +++ b/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