Monitoring an Area for Suspicious OSM Edits Jump to heading#

from __future__ import annotations

import gzip
import logging
import re
import urllib.request
import xml.etree.ElementTree as ET
from collections.abc import Iterable, Iterator
from dataclasses import dataclass, field
from datetime import datetime, timezone

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger("osm.changeset.watch")

STREAM = "https://planet.openstreetmap.org/replication/changesets"
USER_AGENT = "area-edit-monitor/1.0 (ops@example.net)"
QUEUE_TARGET = 5          # items per day; raise the bar if it overflows

VAGUE_COMMENT = re.compile(r"^\s*(|\.|update|edit|fix|test|asdf)\s*$", re.I)


@dataclass(frozen=True)
class Box:
    west: float
    south: float
    east: float
    north: float

    def overlaps(self, other: "Box") -> bool:
        return not (other.east < self.west or other.west > self.east
                    or other.north < self.south or other.south > self.north)

    @property
    def area_deg2(self) -> float:
        return max(self.east - self.west, 0) * max(self.north - self.south, 0)


@dataclass
class Changeset:
    id: int
    user: str
    uid: int
    created_at: str
    closed_at: str | None
    comment: str
    editor: str
    box: Box | None
    created: int = 0
    modified: int = 0
    deleted: int = 0


@dataclass
class Score:
    changeset: Changeset
    points: float
    reasons: list[str] = field(default_factory=list)


def fetch(sequence: int, stream: str = STREAM) -> Iterator[Changeset]:
    text = f"{sequence:09d}"
    url = f"{stream}/{text[0:3]}/{text[3:6]}/{text[6:9]}.osm.gz"
    request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    with urllib.request.urlopen(request, timeout=60) as response:
        payload = gzip.decompress(response.read())
    for element in ET.fromstring(payload).findall("changeset"):
        tags = {t.get("k"): t.get("v") for t in element.findall("tag")}
        box = None
        if element.get("min_lon"):
            box = Box(float(element.get("min_lon")), float(element.get("min_lat")),
                      float(element.get("max_lon")), float(element.get("max_lat")))
        yield Changeset(
            id=int(element.get("id")), user=element.get("user", "?"),
            uid=int(element.get("uid", 0)), created_at=element.get("created_at"),
            closed_at=element.get("closed_at"), comment=tags.get("comment", ""),
            editor=tags.get("created_by", "unknown"), box=box,
            created=int(element.get("num_changes", 0)))


def score(changeset: Changeset, watched: Box,
          known_users: set[int], critical_hits: int = 0) -> Score:
    """Bias toward CONSEQUENCE, not intent.

    An accidental mass delete from a misconfigured editor does the same damage
    as a deliberate one and is considerably more common.
    """
    result = Score(changeset, 0.0)

    if changeset.deleted >= 500:
        result.points += 4
        result.reasons.append(f"{changeset.deleted:,} deletions")
    elif changeset.deleted >= 100:
        result.points += 2
        result.reasons.append(f"{changeset.deleted:,} deletions")

    if critical_hits:
        result.points += min(4.0, critical_hits * 0.5)
        result.reasons.append(f"touches {critical_hits} feature(s) we depend on")

    # A new account is NOT suspicious by itself; it only adds weight to a
    # changeset that already looks consequential.
    if changeset.uid not in known_users and result.points > 0:
        result.points += 1
        result.reasons.append("editor not seen in this area before")

    if VAGUE_COMMENT.match(changeset.comment) and result.points > 0:
        result.points += 0.5
        result.reasons.append("no meaningful changeset comment")

    # A wide bounding box for few changes means edits scattered far apart,
    # which is what a careless bulk operation looks like.
    if changeset.box and changeset.box.area_deg2 > watched.area_deg2 * 4:
        result.points += 1.5
        result.reasons.append("extent far larger than the watched area")

    return result


def watch(sequences: Iterable[int], watched: Box, known_users: set[int],
          floor: float = 4.0) -> list[Score]:
    queue: list[Score] = []
    for sequence in sequences:
        for changeset in fetch(sequence):
            if changeset.box is None or not watched.overlaps(changeset.box):
                continue
            scored = score(changeset, watched, known_users)
            if scored.points >= floor:
                queue.append(scored)

    queue.sort(key=lambda s: s.points, reverse=True)
    logger.info("%d changeset(s) above the floor", len(queue))
    for item in queue[:20]:
        logger.info("%5.1f  cs/%d by %s — %s", item.points, item.changeset.id,
                    item.changeset.user, "; ".join(item.reasons))
    return queue


def calibrate(queue: list[Score], days: int, target: int = QUEUE_TARGET) -> float:
    """A queue nobody works through protects nothing. Raise the bar instead
    of expecting more attention."""
    per_day = len(queue) / max(days, 1)
    if per_day <= target:
        return 0.0
    keep = sorted((s.points for s in queue), reverse=True)[:target * days]
    logger.warning("%.1f items/day against a target of %d; suggest floor %.1f",
                   per_day, target, keep[-1] if keep else 0.0)
    return keep[-1] if keep else 0.0


if __name__ == "__main__":
    logger.info("score, rank, and keep the queue short enough to be worked")