Remapping Attacks Against Geo-Indistinguishability

from __future__ import annotations

import math
import random
from dataclasses import dataclass
from typing import Callable, Sequence


@dataclass(frozen=True)
class RemapReport:
    x: float
    y: float
    attempts: int
    remapped: bool


def naive_boundary_remap(
    x: float, y: float, *, inside: Callable[[float, float], bool],
    boundary: Sequence[tuple[float, float]],
) -> RemapReport:
    """The dangerous rule: project an outside point to the nearest boundary vertex.

    Included so its failure can be measured, not because it should be deployed. It
    is a many-to-one map onto a small set, so a report ON the boundary tells the
    adversary that the true location was somewhere in the (large) region that maps
    to that vertex — which is far more informative than the mechanism intended.
    """
    if inside(x, y):
        return RemapReport(x, y, 0, False)
    bx, by = min(boundary, key=lambda p: math.hypot(p[0] - x, p[1] - y))
    return RemapReport(bx, by, 1, True)


def resample_remap(
    sample: Callable[[], tuple[float, float]],
    *,
    inside: Callable[[float, float], bool],
    max_attempts: int = 64,
) -> RemapReport:
    """The safe rule: draw again until the report lands inside the region.

    This yields the mechanism's own distribution CONDITIONED on the region, which
    remains a valid geo-indistinguishable release. Cost is unbounded in principle,
    so the attempt cap matters; on exhaustion the caller must fall back to a
    documented, region-wide rule rather than to the last (outside) draw.
    """
    for attempt in range(1, max_attempts + 1):
        x, y = sample()
        if inside(x, y):
            return RemapReport(x, y, attempt, attempt > 1)
    raise RuntimeError(
        "resampling exhausted: the privacy radius is large relative to the region — "
        "reduce the radius or widen the region rather than falling back to a projection"
    )


def optimal_remap(
    x: float, y: float,
    *,
    candidates: Sequence[tuple[float, float]],
    prior: Sequence[float],
    epsilon: float,
) -> RemapReport:
    """Move the report to the candidate minimising expected error under a prior.

    The remap is a function of the REPORT only — never of the true location — which
    is what keeps it post-processing. Weighting by the prior's posterior for each
    candidate given the observed report is the standard optimal-remap construction.
    """
    if len(candidates) != len(prior):
        raise ValueError("prior must be aligned with candidates")
    best, best_cost = None, float("inf")
    for cx, cy in candidates:
        cost = 0.0
        for (px, py), w in zip(candidates, prior):
            d_report = math.hypot(px - x, py - y)
            posterior = w * math.exp(-epsilon * d_report)
            cost += posterior * math.hypot(px - cx, py - cy)
        if cost < best_cost:
            best, best_cost = (cx, cy), cost
    assert best is not None
    return RemapReport(best[0], best[1], 1, best != (x, y))

def _validate() -> None:
    rng = random.Random(17)

    # A square service region, 4 km on a side, centred on the origin.
    def inside(x: float, y: float) -> bool:
        return abs(x) <= 2_000 and abs(y) <= 2_000

    boundary = [(2_000.0, y) for y in range(-2_000, 2_001, 250)]

    def sample_at(cx: float, cy: float, radius_m: float):
        def draw() -> tuple[float, float]:
            theta = rng.random() * 2 * math.pi
            r = -radius_m / math.log(2) * math.log(rng.random())
            return cx + r * math.cos(theta), cy + r * math.sin(theta)
        return draw

    # 1. Resampling always returns a point inside the region.
    draw = sample_at(1_900.0, 0.0, 400.0)
    reports = [resample_remap(draw, inside=inside) for _ in range(500)]
    assert all(inside(r.x, r.y) for r in reports)

    # 2. Resampling must NOT concentrate mass on the boundary.
    on_edge = sum(1 for r in reports if abs(abs(r.x) - 2_000) < 1.0)
    assert on_edge == 0, "the conditional distribution has no atom at the edge"

    # 3. The naive projection DOES concentrate — this is the attack surface.
    naive = [naive_boundary_remap(*draw(), inside=inside, boundary=boundary)
             for _ in range(500)]
    piled = sum(1 for r in naive if r.remapped)
    distinct = len({(round(r.x), round(r.y)) for r in naive if r.remapped})
    assert piled > 100 and distinct <= len(boundary), (piled, distinct)
    # A report at a boundary vertex is consistent with a large half-plane of truths,
    # so an adversary seeing one learns "outside, in this direction" with certainty.

    # 4. Optimal remapping must move reports toward the prior's mass, not away.
    candidates = [(-1_500.0, 0.0), (0.0, 0.0), (1_500.0, 0.0)]
    prior = [0.1, 0.8, 0.1]
    moved = optimal_remap(1_400.0, 0.0, candidates=candidates, prior=prior,
                          epsilon=math.log(2) / 800.0)
    assert moved.x <= 1_500.0

    # 5. Resampling must fail loudly when the region cannot contain the noise.
    tiny = lambda x, y: abs(x) < 5 and abs(y) < 5
    try:
        resample_remap(sample_at(0.0, 0.0, 5_000.0), inside=tiny, max_attempts=32)
    except RuntimeError as exc:
        assert "resampling exhausted" in str(exc)
    else:
        raise AssertionError("an impossible region must raise, not silently project")

    print("remapping: all assertions passed")


_validate()