cameras: map only working HTTP/MJPEG feeds

Masscan open-554 hosts almost never have a public picture. The map now
defaults to cameras with an http(s) snapshot_url (~132 directory cams).

Ingest probes unauthenticated still/MJPEG paths first and skips the rest,
so dead RTSP-only hosts never land on the map. Pass working=false to see
unverified port-554 hits.

app/camera_preview.py

@@ -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."""
     if not _FFMPEG or not url.lower().startswith("rtsp://"):

app/main.py

@@ -23,7 +23,7 @@ import structlog
 from fastapi import FastAPI, HTTPException, Query
 from fastapi.responses import FileResponse, HTMLResponse
 from fastapi.staticfiles import StaticFiles
-from sqlalchemy import and_, func, select, text
+from sqlalchemy import and_, func, or_, select, text
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from database import async_session, init_extensions
@@ -708,6 +708,12 @@ async def list_cameras(
         description="Bounding box 'min_lon,min_lat,max_lon,max_lat'",
     ),
     source: str | None = Query(None, description="Filter by discovery_source"),
+    working: bool = Query(
+        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.",
+    ),
     limit: int = Query(500, ge=1, le=5000),
 ):
     """Cameras for map display, optionally filtered by geographic bbox."""
@@ -715,6 +721,14 @@ async def list_cameras(
 
     async with async_session() as session:
         stmt = select(cam_table).order_by(cam_table.c.last_seen.desc())
+        if working:
+            stmt = stmt.where(
+                cam_table.c.snapshot_url.isnot(None),
+                or_(
+                    cam_table.c.snapshot_url.startswith("http://"),
+                    cam_table.c.snapshot_url.startswith("https://"),
+                ),
+            )
         if bbox:
             try:
                 min_lon, min_lat, max_lon, max_lat = (

app/masscan_scanner.py

@@ -94,20 +94,36 @@ def extract_open_ips(records: list[dict], port: int) -> list[str]:
 
 # ── Persistence ───────────────────────────────────────────────────────────
 
-async def ingest_open_hosts(ips: list[str]) -> tuple[int, int]:
-    """Insert-or-refresh camera rows for open RTSP hosts.
+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.
 
-    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.
+    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, 0
+        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(list(dict.fromkeys(ips)))
+    coords = await geolocate_ips([ip for ip, _ in live])
     new = 0
     async with async_session() as session:
-        for ip in dict.fromkeys(ips):
+        for ip, feed in live:
             url = build_rtsp_url(ip)
             h = url_hash(url)
             lat, lon = coords.get(ip, (None, None))
@@ -118,7 +134,7 @@ async def ingest_open_hosts(ips: list[str]) -> tuple[int, int]:
                 await session.execute(cameras.insert().values(
                     url_hash=h,
                     source_url=url,
-                    snapshot_url=None,          # RTSP-only; no HTTP snapshot
+                    snapshot_url=feed,
                     discovery_source=MASSCAN_DISCOVERY_SOURCE,
                     location_lat=lat,
                     location_lon=lon,
@@ -127,7 +143,8 @@ async def ingest_open_hosts(ips: list[str]) -> tuple[int, int]:
                     device_type="rtsp",
                     first_seen=now,
                     last_seen=now,
-                    raw={"discovered_via": "masscan", "port": 554},
+                    raw={"discovered_via": "masscan", "port": 554,
+                         "public_feed": feed},
                 ))
                 new += 1
             else:
@@ -135,13 +152,15 @@ async def ingest_open_hosts(ips: list[str]) -> tuple[int, int]:
                     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, %d refreshed", new, len(set(ips)))
-    return new, len(set(ips))
+    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 ──────────────────────────────────────────────────────────
@@ -201,7 +220,7 @@ async def flush(seen: set[str], new_accum: int) -> tuple[int, int]:
     if not seen:
         return 0, 0
     ips = list(seen)
-    new, _ = await ingest_open_hosts(ips)
-    published = await publish_new_hosts(ips)
+    new, live = await ingest_open_hosts(ips)
+    published = await publish_new_hosts(live)
     seen.clear()
     return new, published