377 lines
14 KiB
Python
377 lines
14 KiB
Python
|
"""Camera discovery scraper — finds publicly listed open IP cameras.
|
||
|
|
|
||
|
|
Ethics/scope (hard constraints, enforced in code):
|
||
|
|
* Only indexes cameras that appear in PUBLIC directories/lists.
|
||
|
|
* No credential brute-forcing, no login attempts of any kind.
|
||
|
|
* No active scanning — never probes private (RFC1918/loopback/link-local)
|
||
|
|
ranges; private-range URLs found in public lists are dropped.
|
||
|
|
|
||
|
|
Pipeline per cycle:
|
||
|
|
1. Fetch each configured public source page/list.
|
||
|
|
2. Parse cameras (Insecam-style HTML with embedded lat/lon, or plain-text
|
||
|
|
`url[|lat,lon|vendor|location]` lines).
|
||
|
|
3. Geocode missing coords via Nominatim at <=1 req/s.
|
||
|
|
4. Publish each camera to NATS (`events.camera`) for the shared ingester,
|
||
|
|
and upsert into the `cameras` table keyed by sha256(url_hash).
|
||
|
|
|
||
|
|
Snapshots are cached locally with a TTL; nothing is fetched more often than
|
||
|
|
the cache allows.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import hashlib
|
||
|
|
import ipaddress
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import re
|
||
|
|
import time
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
from urllib.parse import urlparse
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
from camera_config import (
|
||
|
|
CAMERA_SOURCE_URLS, CAMERA_REQUEST_DELAY, CAMERA_MAX_PER_SOURCE,
|
||
|
|
NOMINATIM_URL, NOMINATIM_MIN_INTERVAL, USER_AGENT,
|
||
|
|
SNAPSHOT_CACHE_DIR, SNAPSHOT_TTL_SECONDS, SNAPSHOT_TIMEOUT,
|
||
|
|
)
|
||
|
|
from camera_models import cameras
|
||
|
|
from database import async_session
|
||
|
|
|
||
|
|
logger = logging.getLogger("osint.camera_scraper")
|
||
|
|
|
||
|
|
# ── Private-range guard ────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
def is_public_url(url: str) -> bool:
|
||
|
|
"""True only if the URL host resolves to a globally-routable address.
|
||
|
|
|
||
|
|
Hostnames that cannot be resolved are treated as not-public (fail closed).
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
parsed = urlparse(url)
|
||
|
|
if parsed.scheme not in ("http", "https", "rtsp", "rtsps"):
|
||
|
|
return False
|
||
|
|
host = parsed.hostname
|
||
|
|
if not host:
|
||
|
|
return False
|
||
|
|
# Literal IPs: check directly.
|
||
|
|
try:
|
||
|
|
addr = ipaddress.ip_address(host)
|
||
|
|
return addr.is_global and not addr.is_private
|
||
|
|
except ValueError:
|
||
|
|
pass
|
||
|
|
# Private-name quick rejects before any DNS work.
|
||
|
|
if (host.endswith(".local") or host.endswith(".internal")
|
||
|
|
or host == "localhost" or re.fullmatch(r"10\..*|172\.(1[6-9]|2\d|3[01])\..*|192\.168\..*", host)):
|
||
|
|
return False
|
||
|
|
import socket
|
||
|
|
for info in socket.getaddrinfo(host, None):
|
||
|
|
addr = ipaddress.ip_address(info[4][0])
|
||
|
|
if not (addr.is_global and not addr.is_private):
|
||
|
|
return False
|
||
|
|
return True
|
||
|
|
except Exception: # noqa: BLE001
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
# ── Dedupe key ─────────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
def url_hash(url: str) -> str:
|
||
|
|
return hashlib.sha256(url.strip().lower().encode()).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
# ── Politeness-limited fetcher ─────────────────────────────────────────────
|
||
|
|
|
||
|
|
class RateLimitedClient:
|
||
|
|
"""httpx client wrapper enforcing a per-host minimum request interval."""
|
||
|
|
|
||
|
|
def __init__(self, min_delay: float = CAMERA_REQUEST_DELAY):
|
||
|
|
self._delay = min_delay
|
||
|
|
self._last: dict[str, float] = {}
|
||
|
|
self.client = httpx.AsyncClient(
|
||
|
|
timeout=20, follow_redirects=True,
|
||
|
|
headers={"User-Agent": USER_AGENT},
|
||
|
|
)
|
||
|
|
|
||
|
|
async def get(self, url: str, **kw) -> httpx.Response:
|
||
|
|
host = urlparse(url).netloc
|
||
|
|
now = time.monotonic()
|
||
|
|
wait = self._last.get(host, 0.0) + self._delay - now
|
||
|
|
if wait > 0:
|
||
|
|
await asyncio.sleep(wait)
|
||
|
|
self._last[host] = time.monotonic()
|
||
|
|
return await self.client.get(url, **kw)
|
||
|
|
|
||
|
|
async def aclose(self):
|
||
|
|
await self.client.aclose()
|
||
|
|
|
||
|
|
|
||
|
|
# ── Nominatim geocoding (1 req/s politeness) ───────────────────────────────
|
||
|
|
|
||
|
|
class Geocoder:
|
||
|
|
def __init__(self):
|
||
|
|
self._min_interval = NOMINATIM_MIN_INTERVAL
|
||
|
|
self._last = 0.0
|
||
|
|
self._cache: dict[str, tuple[float | None, float | None]] = {}
|
||
|
|
self._lock = asyncio.Lock()
|
||
|
|
|
||
|
|
async def geocode(self, place: str) -> tuple[float | None, float | None]:
|
||
|
|
place = place.strip()
|
||
|
|
if not place:
|
||
|
|
return None, None
|
||
|
|
if place in self._cache:
|
||
|
|
return self._cache[place]
|
||
|
|
async with self._lock:
|
||
|
|
wait = self._last + self._min_interval - time.monotonic()
|
||
|
|
if wait > 0:
|
||
|
|
await asyncio.sleep(wait)
|
||
|
|
try:
|
||
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
||
|
|
r = await c.get(
|
||
|
|
f"{NOMINATIM_URL}/search",
|
||
|
|
params={"q": place, "format": "json", "limit": 1},
|
||
|
|
headers={"User-Agent": USER_AGENT},
|
||
|
|
)
|
||
|
|
r.raise_for_status()
|
||
|
|
data = r.json()
|
||
|
|
result = (
|
||
|
|
(float(data[0]["lat"]), float(data[0]["lon"]))
|
||
|
|
if data else (None, None)
|
||
|
|
)
|
||
|
|
except Exception: # noqa: BLE001
|
||
|
|
logger.warning("geocode failed for %r", place, exc_info=True)
|
||
|
|
result = (None, None)
|
||
|
|
self._last = time.monotonic()
|
||
|
|
self._cache[place] = result
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
# ── Snapshot cache (local disk, TTL'd) ─────────────────────────────────────
|
||
|
|
|
||
|
|
_snapshot_lock = asyncio.Lock()
|
||
|
|
|
||
|
|
|
||
|
|
async def fetch_snapshot(url: str) -> bytes | None:
|
||
|
|
"""Fetch a snapshot image through the local TTL cache. Returns raw bytes."""
|
||
|
|
key = url_hash(url)[:24]
|
||
|
|
path = Path(SNAPSHOT_CACHE_DIR) / f"{key}.bin"
|
||
|
|
meta = Path(SNAPSHOT_CACHE_DIR) / f"{key}.meta"
|
||
|
|
async with _snapshot_lock:
|
||
|
|
try:
|
||
|
|
if path.exists() and meta.exists():
|
||
|
|
age = time.time() - path.stat().st_mtime
|
||
|
|
if age < SNAPSHOT_TTL_SECONDS:
|
||
|
|
return path.read_bytes()
|
||
|
|
except OSError:
|
||
|
|
pass
|
||
|
|
try:
|
||
|
|
async with httpx.AsyncClient(timeout=SNAPSHOT_TIMEOUT, follow_redirects=True) as c:
|
||
|
|
r = await c.get(url, headers={"User-Agent": USER_AGENT})
|
||
|
|
if r.status_code != 200 or len(r.content) < 64:
|
||
|
|
return None
|
||
|
|
data = r.content
|
||
|
|
except Exception: # noqa: BLE001
|
||
|
|
return None
|
||
|
|
async with _snapshot_lock:
|
||
|
|
try:
|
||
|
|
Path(SNAPSHOT_CACHE_DIR).mkdir(parents=True, exist_ok=True)
|
||
|
|
path.write_bytes(data)
|
||
|
|
meta.write_text(json.dumps({"url": url, "fetched": time.time()}))
|
||
|
|
except OSError:
|
||
|
|
logger.warning("snapshot cache write failed", exc_info=True)
|
||
|
|
return data
|
||
|
|
|
||
|
|
|
||
|
|
# ── Source parsers ─────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
# Insecam-style pages embed camera entries like:
|
||
|
|
# <a href="/view/..."> ... latitude: 48.85 / longitude: 2.35 ... title="Paris"
|
||
|
|
# We parse generically: find lat/lon pairs plus nearby snapshot <img> sources.
|
||
|
|
_LATLON_RE = re.compile(
|
||
|
|
r"(?:latitude|lat)[\"'\s:=]+(-?\d{1,2}\.\d+).{0,200}?(?:longitude|lon)[\"'\s:=]+(-?\d{1,3}\.\d+)",
|
||
|
|
re.I | re.S,
|
||
|
|
)
|
||
|
|
_IMG_RE = re.compile(r"<img[^>]+src=[\"']([^\"']+\.(?:jpg|jpeg|png|mjpeg))[\"']", re.I)
|
||
|
|
_TITLE_RE = re.compile(r"<title>([^<]+)</title>", re.I)
|
||
|
|
|
||
|
|
|
||
|
|
def _vendor_from_url(url: str) -> str | None:
|
||
|
|
u = url.lower()
|
||
|
|
for vendor, marker in [
|
||
|
|
("hikvision", "hikvision"), ("dahua", "dahua"),
|
||
|
|
("axis", "axis-cgi"), ("foscam", "foscam"),
|
||
|
|
("tplink", "tplink"), ("tp-link", "tp-link"),
|
||
|
|
("mjpeg", "mjpeg"), ("rtsp", "rtsp://"),
|
||
|
|
]:
|
||
|
|
if marker in u:
|
||
|
|
return vendor
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def parse_plain_list(text: str, source_name: str) -> list[dict]:
|
||
|
|
"""Parse lines of: url[|lat,lon|vendor|location_name]"""
|
||
|
|
cams = []
|
||
|
|
for line in text.splitlines():
|
||
|
|
line = line.strip()
|
||
|
|
if not line or line.startswith("#"):
|
||
|
|
continue
|
||
|
|
parts = [p.strip() for p in line.split("|")]
|
||
|
|
url = parts[0]
|
||
|
|
lat = lon = vendor = loc = None
|
||
|
|
if len(parts) > 1 and "," in parts[1]:
|
||
|
|
try:
|
||
|
|
la, lo = parts[1].split(",", 1)
|
||
|
|
lat, lon = float(la), float(lo)
|
||
|
|
except ValueError:
|
||
|
|
pass
|
||
|
|
if len(parts) > 2 and parts[2]:
|
||
|
|
vendor = parts[2]
|
||
|
|
if len(parts) > 3 and parts[3]:
|
||
|
|
loc = parts[3]
|
||
|
|
if not url.startswith(("http://", "https://", "rtsp://")):
|
||
|
|
continue
|
||
|
|
cams.append({
|
||
|
|
"source_url": url,
|
||
|
|
"snapshot_url": url,
|
||
|
|
"discovery_source": source_name,
|
||
|
|
"location_lat": lat, "location_lon": lon,
|
||
|
|
"location_name": loc, "vendor": vendor,
|
||
|
|
})
|
||
|
|
return cams
|
||
|
|
|
||
|
|
|
||
|
|
def parse_directory_html(html: str, base_url: str, source_name: str) -> list[dict]:
|
||
|
|
"""Generic Insecam-style directory parser."""
|
||
|
|
cams = []
|
||
|
|
imgs = _IMG_RE.findall(html)
|
||
|
|
for m in _LATLON_RE.finditer(html):
|
||
|
|
lat, lon = float(m.group(1)), float(m.group(2))
|
||
|
|
# nearest img tag after the match → best-effort snapshot endpoint
|
||
|
|
tail = html[m.end(): m.end() + 800]
|
||
|
|
img_m = _IMG_RE.search(tail)
|
||
|
|
snap = img_m.group(1) if img_m else (imgs[0] if imgs else None)
|
||
|
|
if snap and snap.startswith("/"):
|
||
|
|
from urllib.parse import urljoin
|
||
|
|
snap = urljoin(base_url, snap)
|
||
|
|
cams.append({
|
||
|
|
"source_url": base_url,
|
||
|
|
"snapshot_url": snap,
|
||
|
|
"discovery_source": source_name,
|
||
|
|
"location_lat": lat, "location_lon": lon,
|
||
|
|
"location_name": None, "vendor": None,
|
||
|
|
})
|
||
|
|
return cams
|
||
|
|
|
||
|
|
|
||
|
|
async def scrape_source(client: RateLimitedClient, geo: Geocoder,
|
||
|
|
src_url: str) -> list[dict]:
|
||
|
|
"""Fetch one source and return normalized camera dicts (public only)."""
|
||
|
|
try:
|
||
|
|
resp = await client.get(src_url)
|
||
|
|
resp.raise_for_status()
|
||
|
|
except Exception: # noqa: BLE001
|
||
|
|
logger.exception("failed to fetch source %s", src_url)
|
||
|
|
return []
|
||
|
|
|
||
|
|
name = urlparse(src_url).netloc
|
||
|
|
ctype = resp.headers.get("content-type", "")
|
||
|
|
if "html" in ctype:
|
||
|
|
cams = parse_directory_html(resp.text, str(resp.url), name)
|
||
|
|
else:
|
||
|
|
cams = parse_plain_list(resp.text, name)
|
||
|
|
|
||
|
|
out: list[dict] = []
|
||
|
|
seen_in_batch: set[str] = set()
|
||
|
|
for cam in cams[:CAMERA_MAX_PER_SOURCE]:
|
||
|
|
url = cam["source_url"]
|
||
|
|
h = url_hash(url)
|
||
|
|
if h in seen_in_batch:
|
||
|
|
continue
|
||
|
|
seen_in_batch.add(h)
|
||
|
|
# Hard scope guard: drop anything not publicly routable.
|
||
|
|
if not is_public_url(url):
|
||
|
|
continue
|
||
|
|
if cam["location_lat"] is None and cam["location_name"]:
|
||
|
|
cam["location_lat"], cam["location_lon"] = await geo.geocode(
|
||
|
|
cam["location_name"])
|
||
|
|
if cam["location_lat"] is None and cam["location_lon"] is None:
|
||
|
|
continue # map display requires coordinates
|
||
|
|
cam.setdefault("vendor", None)
|
||
|
|
if not cam.get("vendor"):
|
||
|
|
cam["vendor"] = _vendor_from_url(cam["snapshot_url"] or url)
|
||
|
|
cam["device_type"] = "rtsp" if (cam["snapshot_url"] or "").startswith("rtsp") else "ip-cam"
|
||
|
|
cam["raw"] = {"discovered_via": src_url}
|
||
|
|
out.append(cam)
|
||
|
|
logger.info("source %s yielded %d public cameras", src_url, len(out))
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
# ── Persistence ────────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
async def upsert_cameras(cams: list[dict]) -> int:
|
||
|
|
"""Insert-or-update cameras keyed by url_hash. Returns rows written."""
|
||
|
|
now = datetime.now(timezone.utc)
|
||
|
|
written = 0
|
||
|
|
async with async_session() as session:
|
||
|
|
for cam in cams:
|
||
|
|
values = {
|
||
|
|
"url_hash": url_hash(cam["source_url"]),
|
||
|
|
"source_url": cam["source_url"],
|
||
|
|
"snapshot_url": cam.get("snapshot_url"),
|
||
|
|
"discovery_source": cam["discovery_source"],
|
||
|
|
"location_lat": cam.get("location_lat"),
|
||
|
|
"location_lon": cam.get("location_lon"),
|
||
|
|
"location_name": cam.get("location_name"),
|
||
|
|
"vendor": cam.get("vendor"),
|
||
|
|
"device_type": cam.get("device_type"),
|
||
|
|
"last_seen": now,
|
||
|
|
"raw": cam.get("raw"),
|
||
|
|
}
|
||
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||
|
|
stmt = pg_insert(cameras).values(**values)
|
||
|
|
stmt = stmt.on_conflict_do_update(
|
||
|
|
index_elements=[cameras.c.url_hash],
|
||
|
|
set_={
|
||
|
|
"last_seen": now,
|
||
|
|
"snapshot_url": values["snapshot_url"],
|
||
|
|
"location_lat": values["location_lat"],
|
||
|
|
"location_lon": values["location_lon"],
|
||
|
|
"location_name": values["location_name"],
|
||
|
|
"vendor": values["vendor"],
|
||
|
|
"device_type": values["device_type"],
|
||
|
|
"raw": values["raw"],
|
||
|
|
},
|
||
|
|
)
|
||
|
|
await session.execute(stmt)
|
||
|
|
written += 1
|
||
|
|
await session.commit()
|
||
|
|
return written
|
||
|
|
|
||
|
|
|
||
|
|
# ── Cycle driver ───────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
async def run_cycle() -> int:
|
||
|
|
client = RateLimitedClient()
|
||
|
|
geo = Geocoder()
|
||
|
|
total = 0
|
||
|
|
try:
|
||
|
|
results = await asyncio.gather(
|
||
|
|
*(scrape_source(client, geo, s) for s in CAMERA_SOURCE_URLS),
|
||
|
|
return_exceptions=True,
|
||
|
|
)
|
||
|
|
all_cams: list[dict] = []
|
||
|
|
for r in results:
|
||
|
|
if isinstance(r, BaseException):
|
||
|
|
logger.error("scrape task failed: %s", r)
|
||
|
|
else:
|
||
|
|
all_cams.extend(r)
|
||
|
|
if all_cams:
|
||
|
|
total = await upsert_cameras(all_cams)
|
||
|
|
finally:
|
||
|
|
await client.aclose()
|
||
|
|
logger.info("camera cycle complete: %d cameras stored", total)
|
||
|
|
return total
|