From 4aeb747d56d56c9462d4c5c58e634a71b5167fc0 Mon Sep 17 00:00:00 2001 From: AHMED Date: Fri, 9 Oct 2026 15:52:08 +0300 Subject: [PATCH] feat: validate Anwarpy publication metadata --- .gitignore | 4 + automation/anwarpy/page_state.py | 51 +++++++ automation/anwarpy/public-hosts.txt | 1 + automation/anwarpy/publication_meta.py | 155 ++++++++++++++++++++ automation/anwarpy/sanitize.py | 73 +++++++++ automation/anwarpy/test_page_state.py | 50 +++++++ automation/anwarpy/test_publication_meta.py | 79 ++++++++++ automation/anwarpy/test_sanitize.py | 44 ++++++ 8 files changed, 457 insertions(+) create mode 100644 automation/anwarpy/page_state.py create mode 100644 automation/anwarpy/public-hosts.txt create mode 100644 automation/anwarpy/publication_meta.py create mode 100644 automation/anwarpy/sanitize.py create mode 100644 automation/anwarpy/test_page_state.py create mode 100644 automation/anwarpy/test_publication_meta.py create mode 100644 automation/anwarpy/test_sanitize.py diff --git a/.gitignore b/.gitignore index 00d5f4f..2b48746 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,7 @@ screenshot-*.png .DS_Store .vscode/ .idea/ + +# Python runtime files +__pycache__/ +*.pyc diff --git a/automation/anwarpy/page_state.py b/automation/anwarpy/page_state.py new file mode 100644 index 0000000..2fef048 --- /dev/null +++ b/automation/anwarpy/page_state.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +from dataclasses import dataclass +import re +from typing import Iterable, Mapping, Optional + +from publication_meta import clean, parse_meta + + +class MarkerError(ValueError): + pass + + +@dataclass(frozen=True) +class PageMarker: + block_id: str + raw: str + meta: dict[str, str] + + +def find_marker(blocks: Iterable[Mapping[str, object]]) -> Optional[PageMarker]: + valid: list[PageMarker] = [] + near_misses: list[str] = [] + + for block in blocks: + block_id = str(block.get("id", "unknown")) + raw_value = block.get("text", "") + raw = raw_value if isinstance(raw_value, str) else "" + normalized = clean(raw) + if re.search(r"anwarpy", normalized, re.IGNORECASE) is None: + continue + if not normalized.startswith("anwarpy:"): + near_misses.append(block_id) + continue + try: + meta = parse_meta(normalized) + except ValueError: + near_misses.append(block_id) + continue + valid.append(PageMarker(block_id=block_id, raw=raw, meta=meta)) + + if len(valid) > 1: + ids = ", ".join(marker.block_id for marker in valid) + raise MarkerError(f"multiple anwarpy markers found in blocks: {ids}") + if valid and near_misses: + ids = ", ".join(near_misses) + raise MarkerError(f"ambiguous anwarpy markers; near-miss blocks: {ids}") + if near_misses: + ids = ", ".join(near_misses) + raise MarkerError(f"near-miss anwarpy marker in blocks: {ids}") + return valid[0] if valid else None diff --git a/automation/anwarpy/public-hosts.txt b/automation/anwarpy/public-hosts.txt new file mode 100644 index 0000000..d83f0ee --- /dev/null +++ b/automation/anwarpy/public-hosts.txt @@ -0,0 +1 @@ +anwarpy.ai diff --git a/automation/anwarpy/publication_meta.py b/automation/anwarpy/publication_meta.py new file mode 100644 index 0000000..9ae7507 --- /dev/null +++ b/automation/anwarpy/publication_meta.py @@ -0,0 +1,155 @@ +from __future__ import annotations + +from datetime import datetime +import re +import unicodedata +from urllib.parse import urlsplit + +CATEGORIES = ("وكلاء", "نماذج", "استرجاع", "سلامة", "أدوات", "أعمال") +ACCESSES = ("free", "vip") +STATUSES = ("pending", "publishing", "published", "blocked", "republish") +BLOCKED_REASONS = ("sanitizer", "generation", "article_schema", "cover", "video") +SLUG_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") +_INVISIBLE = re.compile(r"[\u061c\u200e\u200f\u202a-\u202e\u2066-\u2069\ufeff]") +_NO_BREAK = re.compile(r"[\u00a0\u202f]") + +_FIELD_ORDER = ( + "status", + "access", + "category", + "queued_at", + "slug", + "published_at", + "url", + "blocked_reason", +) +_REQUIRED = { + "pending": {"status", "access", "category", "queued_at"}, + "publishing": {"status", "access", "category", "queued_at", "slug"}, + "published": {"status", "access", "category", "queued_at", "slug", "published_at", "url"}, + "blocked": {"status", "access", "category", "queued_at", "blocked_reason"}, + "republish": {"status", "access", "category", "queued_at", "slug", "published_at", "url"}, +} + + +def clean(line: str) -> str: + line = _INVISIBLE.sub(" ", line) + line = _NO_BREAK.sub(" ", line) + return unicodedata.normalize("NFC", line).strip() + + +def _aware_datetime(value: str, field: str) -> datetime: + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError(f"{field} must be a valid ISO-8601 datetime") from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise ValueError(f"{field} must include a timezone offset") + return parsed + + +def _validate_url(value: str, slug: str) -> None: + parsed = urlsplit(value) + expected_path = f"/posts/{slug}" + if ( + parsed.scheme != "https" + or parsed.netloc != "anwarpy.ai" + or parsed.path != expected_path + or parsed.query + or parsed.fragment + ): + raise ValueError(f"url must equal https://anwarpy.ai{expected_path}") + + +def parse_meta(raw: str) -> dict[str, str]: + line = clean(raw) + if not line.startswith("anwarpy:"): + raise ValueError("marker must start with exact prefix anwarpy:") + + body = line.removeprefix("anwarpy:").strip() + if not body: + raise ValueError("marker fields are required") + + meta: dict[str, str] = {} + for part in body.split("|"): + segment = clean(part) + if "=" not in segment: + raise ValueError(f"invalid marker field: {segment}") + key, value = (clean(item) for item in segment.split("=", 1)) + if key not in _FIELD_ORDER: + raise ValueError(f"unknown marker field: {key}") + if key in meta: + raise ValueError(f"duplicate marker field: {key}") + if not value: + raise ValueError(f"empty marker field: {key}") + meta[key] = value + + status = meta.get("status") + if status not in STATUSES: + raise ValueError("status is invalid") + required = _REQUIRED[status] + missing = required - meta.keys() + if missing: + raise ValueError(f"missing required fields: {', '.join(sorted(missing))}") + forbidden = meta.keys() - required + if forbidden: + raise ValueError(f"fields not allowed for {status}: {', '.join(sorted(forbidden))}") + + if meta["access"] not in ACCESSES: + raise ValueError("access is invalid") + if meta["category"] not in CATEGORIES: + raise ValueError("category is invalid") + _aware_datetime(meta["queued_at"], "queued_at") + + slug = meta.get("slug") + if slug is not None and not SLUG_RE.fullmatch(slug): + raise ValueError("slug must be lowercase ASCII kebab-case") + if "published_at" in meta: + _aware_datetime(meta["published_at"], "published_at") + if "url" in meta: + _validate_url(meta["url"], meta["slug"]) + if "blocked_reason" in meta and meta["blocked_reason"] not in BLOCKED_REASONS: + raise ValueError("blocked_reason is invalid") + + return meta + + +def render_meta(meta: dict[str, str]) -> str: + validated = parse_meta( + "anwarpy: " + " | ".join(f"{key}={value}" for key, value in meta.items()) + ) + return "anwarpy: " + " | ".join( + f"{key}={validated[key]}" for key in _FIELD_ORDER if key in validated + ) + + +def same_meta(expected_raw: str, actual_raw: str) -> bool: + return parse_meta(expected_raw) == parse_meta(actual_raw) + + +def mark_publishing(raw: str, slug: str) -> str: + meta = parse_meta(raw) + if meta["status"] != "pending": + raise ValueError("only pending markers can be claimed") + meta["status"] = "publishing" + meta["slug"] = clean(slug) + return render_meta(meta) + + +def mark_published(raw: str, published_at: str, url: str) -> str: + meta = parse_meta(raw) + if meta["status"] not in ("publishing", "republish"): + raise ValueError("only publishing or republish markers can become published") + meta["status"] = "published" + meta["published_at"] = clean(published_at) + meta["url"] = clean(url) + return render_meta(meta) + + +def mark_blocked(raw: str, reason: str) -> str: + meta = parse_meta(raw) + if meta["status"] != "pending": + raise ValueError("only pending markers can be blocked") + meta["status"] = "blocked" + meta["blocked_reason"] = clean(reason) + return render_meta(meta) diff --git a/automation/anwarpy/sanitize.py b/automation/anwarpy/sanitize.py new file mode 100644 index 0000000..cfa00b3 --- /dev/null +++ b/automation/anwarpy/sanitize.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import ipaddress +import re +from typing import Iterable +from urllib.parse import urlsplit + +REDACTION = "[REDACTED]" + +_URL_RE = re.compile(r"https?://[^\s<>\]\[\)\(\"']+", re.IGNORECASE) +_EMAIL_RE = re.compile(r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b", re.IGNORECASE) +_IPV4_RE = re.compile(r"(? set[str]: + return {host.strip().lower().rstrip(".") for host in hosts if host.strip()} + + +def _redact_url(match: re.Match[str], allowed_hosts: set[str]) -> str: + value = match.group(0) + host = (urlsplit(value).hostname or "").lower().rstrip(".") + return value if host in allowed_hosts else REDACTION + + +def _redact_ipv4(match: re.Match[str]) -> str: + value = match.group(0) + try: + ipaddress.IPv4Address(value) + except ipaddress.AddressValueError: + return value + return REDACTION + + +def _redact_ipv6(match: re.Match[str]) -> str: + value = match.group(0) + try: + ipaddress.IPv6Address(value) + except ipaddress.AddressValueError: + return value + return REDACTION + + +def _redact_fqdn(match: re.Match[str], allowed_hosts: set[str]) -> str: + host = match.group(0) + return host if host.lower().rstrip(".") in allowed_hosts else REDACTION + + +def sanitize_text(text: str, *, allowed_hosts: Iterable[str]) -> str: + hosts = _normalized_hosts(allowed_hosts) + cleaned = _URL_RE.sub(lambda match: _redact_url(match, hosts), text) + cleaned = _CREDENTIAL_RE.sub(REDACTION, cleaned) + cleaned = _EMAIL_RE.sub(REDACTION, cleaned) + cleaned = _PATH_RE.sub(REDACTION, cleaned) + cleaned = _IPV4_RE.sub(_redact_ipv4, cleaned) + cleaned = _IPV6_CANDIDATE_RE.sub(_redact_ipv6, cleaned) + cleaned = _PHONE_RE.sub(REDACTION, cleaned) + cleaned = _PRIVATE_HOST_RE.sub(REDACTION, cleaned) + cleaned = _FQDN_RE.sub(lambda match: _redact_fqdn(match, hosts), cleaned) + return cleaned + + +def assert_publishable(text: str, *, allowed_hosts: Iterable[str]) -> None: + sanitized = sanitize_text(text, allowed_hosts=allowed_hosts) + if sanitized != text: + raise ValueError("content still contains private or unapproved data") diff --git a/automation/anwarpy/test_page_state.py b/automation/anwarpy/test_page_state.py new file mode 100644 index 0000000..144da79 --- /dev/null +++ b/automation/anwarpy/test_page_state.py @@ -0,0 +1,50 @@ +import unittest + +from page_state import MarkerError, find_marker + + +VALID = "anwarpy: status=pending | access=free | category=نماذج | queued_at=2026-10-09T15:12:00+03:00" + + +class PageStateTests(unittest.TestCase): + def test_finds_one_exact_marker_after_unrelated_blocks(self): + result = find_marker([ + {"id": "a", "text": "ملاحظات"}, + {"id": "b", "text": VALID}, + {"id": "c", "text": "مصدر عام"}, + ]) + self.assertEqual(result.block_id, "b") + self.assertEqual(result.meta["status"], "pending") + + def test_returns_none_when_anwarpy_is_never_mentioned(self): + self.assertIsNone(find_marker([{"id": "a", "text": "لا توجد علامة"}])) + + def test_reports_every_near_miss(self): + for text in ( + VALID.replace("anwarpy:", "Anwarpy:"), + "prefix " + VALID, + "documentation mentions anwarpy but has no marker", + VALID.replace("anwarpy:", "anwarpy ="), + ): + with self.subTest(text=text): + with self.assertRaisesRegex(MarkerError, "near-miss"): + find_marker([{"id": "bad", "text": text}]) + + def test_scans_all_blocks_and_rejects_ambiguity(self): + with self.assertRaisesRegex(MarkerError, "multiple"): + find_marker([ + {"id": "first", "text": VALID}, + {"id": "middle", "text": "unrelated"}, + {"id": "last", "text": VALID.replace("category=نماذج", "category=أدوات")}, + ]) + + def test_exact_plus_near_miss_is_ambiguous(self): + with self.assertRaisesRegex(MarkerError, "ambiguous"): + find_marker([ + {"id": "good", "text": VALID}, + {"id": "bad", "text": "Anwarpy: pending"}, + ]) + + +if __name__ == "__main__": + unittest.main() diff --git a/automation/anwarpy/test_publication_meta.py b/automation/anwarpy/test_publication_meta.py new file mode 100644 index 0000000..80e7909 --- /dev/null +++ b/automation/anwarpy/test_publication_meta.py @@ -0,0 +1,79 @@ +import unittest + +from publication_meta import ( + clean, + mark_blocked, + mark_published, + mark_publishing, + parse_meta, + same_meta, +) + + +PENDING = "anwarpy: status=pending | access=vip | category=أدوات | queued_at=2026-10-09T15:12:00+03:00" + + +class PublicationMetaTests(unittest.TestCase): + def test_parses_pending_and_normalizes_formatting_controls(self): + raw = "\ufeffanwarpy:\u00a0status=pending | access=vip | category=أدوات | queued_at=2026-10-09T15:12:00+03:00\u200f" + meta = parse_meta(raw) + self.assertEqual(meta["status"], "pending") + self.assertEqual(meta["category"], "أدوات") + + def test_rejects_invalid_values_and_dates(self): + for raw in ( + PENDING.replace("access=vip", "access=paid"), + PENDING.replace("category=أدوات", "category=عام"), + PENDING.replace("2026-10-09", "2026-13-45"), + PENDING.replace("+03:00", ""), + ): + with self.subTest(raw=raw): + with self.assertRaises(ValueError): + parse_meta(raw) + + def test_rejects_invalid_state_shapes(self): + with self.assertRaisesRegex(ValueError, "slug"): + parse_meta(PENDING.replace("status=pending", "status=publishing")) + with self.assertRaisesRegex(ValueError, "blocked_reason"): + parse_meta(PENDING.replace("status=pending", "status=blocked")) + with self.assertRaisesRegex(ValueError, "not allowed"): + parse_meta(PENDING + " | url=https://anwarpy.ai/posts/example") + + def test_transitions_reparse_raw_input(self): + publishing = mark_publishing(PENDING, "arabic-ai-guide") + self.assertEqual(parse_meta(publishing)["status"], "publishing") + published = mark_published( + publishing, + "2026-10-09T12:20:11+00:00", + "https://anwarpy.ai/posts/arabic-ai-guide", + ) + self.assertEqual(parse_meta(published)["status"], "published") + + def test_blocked_is_terminal_and_carries_a_bounded_reason(self): + blocked = mark_blocked(PENDING, "cover") + self.assertEqual(parse_meta(blocked)["blocked_reason"], "cover") + with self.assertRaises(ValueError): + mark_blocked(PENDING, "network") + + def test_canonical_url_and_slug_are_exact(self): + publishing = mark_publishing(PENDING, "arabic-ai-guide") + for url in ( + "http://anwarpy.ai/posts/arabic-ai-guide", + "https://www.anwarpy.ai/posts/arabic-ai-guide", + "https://anwarpy.ai/posts/arabic-ai-guide/", + "https://anwarpy.ai/posts/other", + ): + with self.subTest(url=url): + with self.assertRaises(ValueError): + mark_published(publishing, "2026-10-09T15:20:11+03:00", url) + with self.assertRaises(ValueError): + mark_publishing(PENDING, "عنوان-عربي") + + def test_readback_comparison_is_semantic_after_cleaning(self): + formatted = "\u200fanwarpy: status=pending|access=vip|category=أدوات|queued_at=2026-10-09T15:12:00+03:00" + self.assertTrue(same_meta(PENDING, formatted)) + self.assertEqual(clean("\ufeff أداة\u00a0"), "أداة") + + +if __name__ == "__main__": + unittest.main() diff --git a/automation/anwarpy/test_sanitize.py b/automation/anwarpy/test_sanitize.py new file mode 100644 index 0000000..b1d04f3 --- /dev/null +++ b/automation/anwarpy/test_sanitize.py @@ -0,0 +1,44 @@ +import unittest + +from sanitize import assert_publishable, sanitize_text + + +class SanitizeTests(unittest.TestCase): + def test_redacts_private_identifiers_and_credentials(self): + raw = """Connect to 10.1.2.3 or fd00::12 as admin@example.test. +SSH root@private.internal and use password=FakeSecret123. +Read /Users/example/private/config and call https://hidden.internal/api. +Phone +1 555 123 4567; token: abcdefghijklmnop. +""" + cleaned = sanitize_text(raw, allowed_hosts={"anwarpy.ai"}) + for secret in ( + "10.1.2.3", + "fd00::12", + "admin@example.test", + "root@private.internal", + "FakeSecret123", + "/Users/example/private/config", + "hidden.internal", + "+1 555 123 4567", + "abcdefghijklmnop", + ): + self.assertNotIn(secret, cleaned) + assert_publishable(cleaned, allowed_hosts={"anwarpy.ai"}) + + def test_preserves_approved_public_urls(self): + text = "راجع https://anwarpy.ai/posts/example للمزيد." + self.assertEqual(sanitize_text(text, allowed_hosts={"anwarpy.ai"}), text) + assert_publishable(text, allowed_hosts={"anwarpy.ai"}) + + def test_redacts_unapproved_public_hosts(self): + cleaned = sanitize_text("Source: https://unapproved.example/article", allowed_hosts={"anwarpy.ai"}) + self.assertNotIn("unapproved.example", cleaned) + assert_publishable(cleaned, allowed_hosts={"anwarpy.ai"}) + + def test_assert_publishable_raises_without_using_assert_statement(self): + with self.assertRaises(ValueError): + assert_publishable("internal address 192.168.1.10", allowed_hosts={"anwarpy.ai"}) + + +if __name__ == "__main__": + unittest.main()