diff --git a/app/camera_preview.py b/app/camera_preview.py index cbf2f94..426caa5 100644 --- a/app/camera_preview.py +++ b/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://"): diff --git a/app/main.py b/app/main.py index 8691df9..472d841 100644 --- a/app/main.py +++ b/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 = ( diff --git a/app/masscan_scanner.py b/app/masscan_scanner.py index 347a714..3aa287a 100644 --- a/app/masscan_scanner.py +++ b/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