skills/magnus919_agent-skills
1
0
Fork 0
mirror of https://github.com/magnus919/agent-skills.git synced 2026-09-11 19:47:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(raleigh): enforce publication request contracts

Browse Source
This commit is contained in:
Magnus Hedemark
2026-07-26 18:56:09 -04:00
committed by username
parent cc26b4cb27
commit 1bd7b06112
4 changed files with 86 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
raleigh/EVIDENCE-LEDGER.md
+3 -3
View File
@@ -170,12 +170,12 @@ The following public service boundaries were exercised successfully on 2026-07-2
### Verification
- `PYTHONDONTWRITEBYTECODE=1 python3 -m unittest raleigh/tests/test_raleigh.py`: **364 tests passed**.
- `PYTHONDONTWRITEBYTECODE=1 python3 -m unittest raleigh/tests/test_raleigh.py`: **368 tests passed**.
- `python3 -m ruff check raleigh/scripts/raleighlib/core.py raleigh/scripts/raleighlib/public_safety_stats.py raleigh/scripts/raleighlib/cli.py`: passed.
- `python3 scripts/validate-evals.py raleigh`: **11 eval manifests validated**.
- `ruby scripts/validate-skills.rb`: **110 canonical skills validated**.
- `ruby scripts/validate-skill-quality.rb --base HEAD`: **1 changed skill, 0 errors, 0 warnings**.
- `python3 scripts/eval-coverage.py --modified-from HEAD`: ratchet passed; Raleigh remains schema-valid.
- `ruby scripts/validate-skill-quality.rb --base origin/main`: **1 changed skill, 0 errors, 0 warnings**.
- `python3 scripts/eval-coverage.py --modified-from origin/main`: ratchet passed; Raleigh remains schema-valid.
- Live `police reports --year 2025 --quarter 4 --json`: returned the official `Q4 stats` label and canonical government-cloud PDF URL after an availability probe.
- Live `police stats --year 2025 --json`: returned the annual and quarterly publication index with an explicit document-only warning and no fabricated totals.
- Live `fire stats --year 2026 --json`: returned seven official published categories, including medical `7,882`, source revision/retrieval metadata, the annual document URL, and the aggregate-only privacy warning.
raleigh/scripts/raleighlib/core.py
+18 -2
View File
@@ -130,6 +130,9 @@ class AllowlistRedirectHandler(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
if not is_allowed_host(newurl):
raise SecurityError(f"Redirect led to a non-allowlisted host: {newurl}")
final_url_validator = getattr(req, "_raleigh_final_url_validator", None)
if final_url_validator is not None:
final_url_validator(newurl)
# Strip sensitive headers when crossing origins.
old_origin = _origin(req.full_url)
new_origin = _origin(newurl)
@@ -153,7 +156,7 @@ class AllowlistRedirectHandler(urllib.request.HTTPRedirectHandler):
if data is not None and old_origin != new_origin:
raise SecurityError("Cross-origin redirects cannot preserve a request body")
_enforce_method_policy(method, newurl)
return urllib.request.Request(
redirected = urllib.request.Request(
newurl,
headers=new_headers,
method=method,
@@ -161,6 +164,9 @@ class AllowlistRedirectHandler(urllib.request.HTTPRedirectHandler):
origin_req_host=req.origin_req_host,
unverifiable=True,
)
if final_url_validator is not None:
setattr(redirected, "_raleigh_final_url_validator", final_url_validator)
return redirected
# Prebuilt opener with the bounded allowlisted redirect handler.
@@ -317,6 +323,8 @@ def json_request(
req = urllib.request.Request(
url, headers=req_headers, method=effective_method, data=data
)
if final_url_validator is not None:
setattr(req, "_raleigh_final_url_validator", final_url_validator)
with _OPENER.open(req, timeout=timeout or _get_timeout()) as resp:
final_url = resp.geturl()
if not is_allowed_host(final_url):
@@ -356,7 +364,11 @@ def raw_request(
return _read_limited(resp, max_bytes)
def probe_url(url: str, timeout: int | None = None) -> str:
def probe_url(
url: str,
timeout: int | None = None,
final_url_validator: Callable[[str], None] | None = None,
) -> str:
"""Verify that an allowlisted HTTPS resource is available without reading it."""
if not is_allowed_host(url):
raise SecurityError(f"URL host is not allowlisted: {url}")
@@ -364,10 +376,14 @@ def probe_url(url: str, timeout: int | None = None) -> str:
request = urllib.request.Request(
url, headers=_request_headers(), method="HEAD"
)
if final_url_validator is not None:
setattr(request, "_raleigh_final_url_validator", final_url_validator)
with _OPENER.open(request, timeout=timeout or _get_timeout()) as response:
final_url = response.geturl()
if not is_allowed_host(final_url):
raise SecurityError(f"Redirect led to a non-allowlisted host: {final_url}")
if final_url_validator is not None:
final_url_validator(final_url)
return final_url
raleigh/scripts/raleighlib/public_safety_stats.py
+20 -4
View File
@@ -179,6 +179,15 @@ def _fetch_page(agency: str) -> tuple[dict[str, Any], dict[str, str]]:
or path.get("alias") != urllib.parse.urlparse(source["page_url"]).path
):
raise PublishedStatisticsError("published statistics page identity changed")
changed = attrs.get("changed")
if not isinstance(changed, str) or not changed.strip():
raise PublishedStatisticsError("published statistics page revision timestamp is missing")
try:
changed_at = datetime.fromisoformat(changed.replace("Z", "+00:00"))
except ValueError as exc:
raise PublishedStatisticsError("published statistics page revision timestamp is invalid") from exc
if changed_at.tzinfo is None:
raise PublishedStatisticsError("published statistics page revision timestamp has no timezone")
relationships = data.get("relationships")
if not isinstance(relationships, dict):
@@ -233,7 +242,7 @@ def _fetch_page(agency: str) -> tuple[dict[str, Any], dict[str, str]]:
metadata = {
"url": source["page_url"],
"retrieved_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"published_changed_at": str(attrs.get("changed") or ""),
"published_changed_at": changed,
}
return {"fragments": fragments}, metadata
@@ -293,6 +302,8 @@ def _parse_fire(fragments: dict[str, str], page_url: str) -> tuple[list[dict[str
quarterly_parser = _FragmentParser()
quarterly_parser.feed(quarterly_html)
for link in quarterly_parser.links:
if not link["text"]:
raise PublishedStatisticsError("fire quarterly report publication label is missing")
url = _document_url(link["href"], page_url, "fire")
period = re.search(r"(?:^|[-_/])q([1-4])-(20\d{2})(?:[-_/]|$)", url, re.IGNORECASE)
if period is None:
@@ -326,8 +337,10 @@ def _parse_fire(fragments: dict[str, str], page_url: str) -> tuple[list[dict[str
raise PublishedStatisticsError("fire statistics table headers changed")
values = []
for row in rows[1:]:
if len(row) != 2 or not row[0]:
raise PublishedStatisticsError("fire statistics table contains a malformed row")
valid_number = re.fullmatch(r"(?:0|[1-9]\d*|[1-9]\d{0,2}(?:,\d{3})+)", row[1])
if len(row) != 2 or not row[0] or valid_number is None:
if valid_number is None:
raise PublishedStatisticsError("fire statistics table contains a malformed row")
values.append({"label": row[0], "value": int(row[1].replace(",", "")), "published_value": row[1]})
datasets.append({"year": int(match.group(1)), "kind": "incident_totals", "values": values})
@@ -370,9 +383,12 @@ def _assert_available(items: list[dict[str, Any]]) -> None:
raise PublishedStatisticsError(f"publication selection exceeded the {MAX_PUBLISHED_REPORTS}-report limit")
unique = {(item["document_url"], item["agency"]) for item in items}
for url, agency in unique:
try:
final_url = core.probe_url(url)
def validate_final_url(final_url: str) -> None:
_document_url(final_url, url, agency)
try:
final_url = core.probe_url(url, final_url_validator=validate_final_url)
validate_final_url(final_url)
except (
core.SecurityError,
urllib.error.HTTPError,
raleigh/tests/test_raleigh.py
+45 -1
View File
@@ -107,6 +107,18 @@ class CoreTests(unittest.TestCase):
)
self.assertEqual(redirected.get_method(), "HEAD")
def test_redirect_runs_request_specific_validator_before_following(self):
source = "https://raleighnc.gov/jsonapi/node/service/example"
destination = "https://data.raleighnc.gov/other"
request = urllib.request.Request(source, method="GET")
validator = MagicMock(side_effect=core.SecurityError("wrong endpoint"))
setattr(request, "_raleigh_final_url_validator", validator)
with self.assertRaisesRegex(core.SecurityError, "wrong endpoint"):
core.AllowlistRedirectHandler().redirect_request(
request, None, 302, "redirect", {}, destination
)
validator.assert_called_once_with(destination)
def test_cache_read_write_roundtrip(self):
with tempfile.TemporaryDirectory() as tmp:
os.environ["RALEIGH_CACHE"] = tmp
@@ -3019,7 +3031,9 @@ class PublishedPublicSafetyStatisticsTests(unittest.TestCase):
def setUp(self):
self.probe_patcher = patch("raleighlib.public_safety_stats.core.probe_url")
self.probe = self.probe_patcher.start()
self.probe.side_effect = lambda url: url
self.probe.side_effect = lambda url, **kwargs: (
kwargs.get("final_url_validator", lambda value: None)(url) or url
)
self.addCleanup(self.probe_patcher.stop)
def _fixture(self, agency: str):
@@ -3086,6 +3100,15 @@ class PublishedPublicSafetyStatisticsTests(unittest.TestCase):
with self.assertRaisesRegex(public_safety_stats.PublishedStatisticsError, message):
public_safety_stats.reports("police")
def test_invalid_jsonapi_revision_timestamp_fails_visibly(self):
for value, message in ((None, "timestamp is missing"), ("not-a-date", "timestamp is invalid")):
with self.subTest(value=value):
fixture = self._fixture("police")
fixture["data"]["attributes"]["changed"] = value
with patch("raleighlib.public_safety_stats.core.json_request", return_value=fixture):
with self.assertRaisesRegex(public_safety_stats.PublishedStatisticsError, message):
public_safety_stats.reports("police")
def test_malformed_jsonapi_relationship_identifier_fails_visibly(self):
fixture = self._fixture("police")
relationship = fixture["data"]["relationships"]["field_content_primary"]["data"][0]
@@ -3228,6 +3251,27 @@ class PublishedPublicSafetyStatisticsTests(unittest.TestCase):
with self.assertRaisesRegex(public_safety_stats.PublishedStatisticsError, "headers changed"):
public_safety_stats.statistics("fire", 2026)
def test_one_cell_fire_statistics_row_fails_visibly(self):
fixture = self._fixture("fire")
formatted = fixture["included"][0]["attributes"]["field_stories_text_formatted"]
formatted["value"] = formatted["value"].replace(
"<tr><td><strong>Fire</strong></td><td>401</td></tr>",
"<tr><td><strong>Fire</strong></td></tr>",
)
with patch("raleighlib.public_safety_stats.core.json_request", return_value=fixture):
with self.assertRaisesRegex(public_safety_stats.PublishedStatisticsError, "malformed row"):
public_safety_stats.statistics("fire", 2026)
def test_empty_fire_quarterly_label_fails_visibly(self):
fixture = self._fixture("fire")
formatted = fixture["included"][2]["attributes"]["field_stories_text_formatted"]
formatted["value"] = formatted["value"].replace(
">fire statistics from the previous quarter</a>", "></a>"
)
with patch("raleighlib.public_safety_stats.core.json_request", return_value=fixture):
with self.assertRaisesRegex(public_safety_stats.PublishedStatisticsError, "publication label is missing"):
public_safety_stats.reports("fire")
def test_malformed_published_thousands_separator_fails_visibly(self):
fixture = self._fixture("fire")
formatted = fixture["included"][0]["attributes"]["field_stories_text_formatted"]