From c2a363d74208e963c4ddc4810006459c222f2c49 Mon Sep 17 00:00:00 2001 From: Huseyin Kutsi Balci <262747299+kutsibalci@users.noreply.github.com> Date: Sun, 27 Sep 2026 15:13:08 +0300 Subject: [PATCH 1/2] Remove the client code for APIs the product never calls The README already said HeatShield deliberately does not call SIM/Device Swap, the KYC family or Number Verification, yet the client still carried them as portfolio-era leftovers. Removed: - NacClient: 17 methods (number recycling, call forwarding x2, KYC tenure, age and match, number verification, SIM swap x2, device swap x2, consent, roaming, reachability and roaming subscriptions) and their path-table entries, plus the unused congestion_subs path. - FixtureBackend: the matching responses, their helpers and 15 profile fields no remaining API reads. - Simulator: the matching routes and the passthrough bearer check. - tests: the three tests of removed APIs. The simulator round-trip test now exercises the six product APIs over HTTP. Test numbers moved from real Turkish mobile format to the non-assignable +99999 range the README promises. tools/verify_live.py now probes exactly the six APIs: - The broken portfolio table patcher is gone; it looked for heatshield/ inside heatshield/ and for a master table not in the repo. - So is the OAuth fetch, since no Bearer path remains. - It loads .env through tools/_common and defaults to the documented +99999000001000 sandbox number. - tests/test_verify_live.py runs its probes, cleanup and report offline against the fixture backend. It never calls main(), so it never reads .env and makes no network calls. evidence/ is untouched. The 9 Sep 2026 probe can be re-run exactly from tag v0.1-hackathon. The full six-scenario demo JSON is byte-identical; 197 tests pass. Co-Authored-By: Claude Opus 5.5 --- apps/simulator/main.py | 91 ------ packages/nac_client/client.py | 108 ------- packages/nac_client/fixtures.py | 159 +--------- packages/nac_client/privacy.py | 2 +- tests/test_nac_client.py | 65 ++-- tests/test_verify_live.py | 51 ++++ tools/verify_live.py | 510 ++++++-------------------------- 7 files changed, 176 insertions(+), 810 deletions(-) create mode 100644 tests/test_verify_live.py diff --git a/apps/simulator/main.py b/apps/simulator/main.py index e64860a..4ef80b4 100644 --- a/apps/simulator/main.py +++ b/apps/simulator/main.py @@ -55,90 +55,6 @@ def _phone(body: dict) -> str: return p -def _require_bearer(request: Request): - # Passthrough APIs require Bearer in live; the simulator only warns (mandatory with SIM_STRICT_AUTH=1). - if os.environ.get("SIM_STRICT_AUTH") == "1" and not request.headers.get("authorization", "").startswith("Bearer "): - raise HTTPException(401, {"code": "UNAUTHENTICATED", "message": "a Bearer token is required"}) - - -PT = "/passthrough/camara/v1" - - -@app.post(f"{PT}/number-recycling/number-recycling/v0.2/check") -def number_recycling(request: Request, body: dict = Body(...)): - _require_bearer(request) - return fx.number_recycling(_phone(body), body.get("specifiedDate")) - - -@app.post(f"{PT}/call-forwarding-signal/call-forwarding-signal/v0.3/call-forwardings") -def call_forwardings(request: Request, body: dict = Body(...)): - _require_bearer(request) - return fx.call_forwardings(_phone(body)) - - -@app.post(f"{PT}/call-forwarding-signal/call-forwarding-signal/v0.3/unconditional-call-forwardings") -def unconditional_cf(request: Request, body: dict = Body(...)): - _require_bearer(request) - return fx.unconditional_call_forwarding(_phone(body)) - - -@app.post(f"{PT}/kyc-tenure/kyc-tenure/v0.1/check-tenure") -def kyc_tenure(request: Request, body: dict = Body(...)): - _require_bearer(request) - return fx.kyc_tenure(_phone(body), body.get("tenureDate")) - - -def _snake_fields(body: dict, exclude=("phoneNumber", "ageThreshold", "includeContentLock", "includeParentalControl")) -> dict: - import re - return {re.sub(r"(? Any: def health(self) -> dict: return {"mode": self.cfg.mode, "base_url": self.cfg.base_url if self.cfg.mode != "fixture" else None, "breaker": self.breaker.snapshot(), "recent_calls": len(self.calls)} - # ---------------------------------------------------------------- identity / fraud - def number_recycling(self, phone: str, specified_date: str) -> NacResult: - p = normalize_phone(phone) - return self._call("number-recycling", "POST", P["number_recycling"], json={"phoneNumber": p, "specifiedDate": specified_date}, phone=p, - fixture_fn=lambda: self.fx.number_recycling(p, specified_date)) - - def call_forwardings(self, phone: str) -> NacResult: - p = normalize_phone(phone) - return self._call("call-forwarding-signal", "POST", P["call_forwardings"], json={"phoneNumber": p}, phone=p, - fixture_fn=lambda: self.fx.call_forwardings(p)) - - def unconditional_call_forwarding(self, phone: str) -> NacResult: - p = normalize_phone(phone) - return self._call("call-forwarding-signal", "POST", P["unconditional_cf"], json={"phoneNumber": p}, phone=p, - fixture_fn=lambda: self.fx.unconditional_call_forwarding(p)) - - def kyc_tenure(self, phone: str, tenure_date: str) -> NacResult: - p = normalize_phone(phone) - return self._call("kyc-tenure", "POST", P["kyc_tenure"], json={"phoneNumber": p, "tenureDate": tenure_date}, phone=p, - fixture_fn=lambda: self.fx.kyc_tenure(p, tenure_date)) - - def kyc_age(self, phone: str, age_threshold: int, **fields) -> NacResult: - p = normalize_phone(phone) - body = {"phoneNumber": p, "ageThreshold": age_threshold, **{_camel(k): v for k, v in fields.items() if v is not None}} - return self._call("kyc-age-verification", "POST", P["kyc_age"], json=body, phone=p, - fixture_fn=lambda: self.fx.kyc_age(p, age_threshold, **fields)) - - def kyc_match(self, phone: str, **fields) -> NacResult: - p = normalize_phone(phone) - body = {"phoneNumber": p, **{_camel(k): v for k, v in fields.items() if v is not None}} - return self._call("kyc-match", "POST", P["kyc_match"], json=body, phone=p, - fixture_fn=lambda: self.fx.kyc_match(p, **fields)) - - def number_verify(self, phone: str) -> NacResult: - """In live use: called with the OIDC token the device obtains over mobile data (3-legged). SIMULATOR/fixture in the prototype.""" - p = normalize_phone(phone) - return self._call("number-verification", "POST", P["number_verify"], json={"phoneNumber": p}, phone=p, - fixture_fn=lambda: self.fx.number_verify(p)) - - def sim_swap_check(self, phone: str, max_age_h: int | None = None) -> NacResult: - p = normalize_phone(phone) - body = {"phoneNumber": p, **({"maxAge": max_age_h} if max_age_h is not None else {})} - return self._call("sim-swap", "POST", P["sim_swap_check"], json=body, phone=p, fixture_fn=lambda: self.fx.sim_swap_check(p, max_age_h)) - - def sim_swap_date(self, phone: str) -> NacResult: - p = normalize_phone(phone) - return self._call("sim-swap", "POST", P["sim_swap_date"], json={"phoneNumber": p}, phone=p, fixture_fn=lambda: self.fx.sim_swap_date(p)) - - def device_swap_check(self, phone: str, max_age_h: int | None = None) -> NacResult: - p = normalize_phone(phone) - body = {"phoneNumber": p, **({"maxAge": max_age_h} if max_age_h is not None else {})} - return self._call("device-swap", "POST", P["device_swap_check"], json=body, phone=p, fixture_fn=lambda: self.fx.device_swap_check(p, max_age_h)) - - def device_swap_date(self, phone: str) -> NacResult: - p = normalize_phone(phone) - return self._call("device-swap", "POST", P["device_swap_date"], json={"phoneNumber": p}, phone=p, fixture_fn=lambda: self.fx.device_swap_date(p)) - - def consent(self, phone: str, scopes: list[str], purpose: str) -> NacResult: - p = normalize_phone(phone) - return self._call("consent-info", "POST", P["consent"], json={"phoneNumber": p, "scopes": scopes, "purpose": purpose}, phone=p, - fixture_fn=lambda: self.fx.consent(p, scopes, purpose)) - # ---------------------------------------------------------------- device / location def reachability(self, phone: str) -> NacResult: p = normalize_phone(phone) return self._call("device-reachability-status", "POST", P["reachability"], json={"device": {"phoneNumber": p}}, phone=p, fixture_fn=lambda: self.fx.reachability(p)) - def roaming(self, phone: str) -> NacResult: - p = normalize_phone(phone) - return self._call("device-roaming-status", "POST", P["roaming"], json={"device": {"phoneNumber": p}}, phone=p, fixture_fn=lambda: self.fx.roaming(p)) - def location_retrieve(self, phone: str, max_age_s: int | None = None) -> NacResult: p = normalize_phone(phone) body = {"device": {"phoneNumber": p}, **({"maxAge": max_age_s} if max_age_s is not None else {})} @@ -292,31 +209,6 @@ def geofence_get(self, subscription_id: str) -> NacResult: def geofence_delete(self, subscription_id: str) -> NacResult: return self._call("geofencing-subscriptions", "DELETE", f"{P['geofencing_subs']}/{subscription_id}", fixture_fn=lambda: self.fx.delete_subscription(subscription_id)) - def reachability_subscribe(self, phone: str, sink: str, *, types: list[str] | None = None, sink_token: str | None = None, expire: datetime | None = None, initial_event: bool = True) -> NacResult: - p = normalize_phone(phone) - types = types or [ - "org.camaraproject.device-reachability-status-subscriptions.v0.reachability-data", - "org.camaraproject.device-reachability-status-subscriptions.v0.reachability-disconnected", - ] - body = self._sub_body(p, sink, types, sink_token, expire, None, initial_event) - return self._call("device-reachability-status-subscriptions", "POST", P["reachability_subs"], json=body, phone=p, fixture_fn=lambda: self.fx.create_subscription("reachability", body)) - - def reachability_unsubscribe(self, subscription_id: str) -> NacResult: - return self._call("device-reachability-status-subscriptions", "DELETE", f"{P['reachability_subs']}/{subscription_id}", fixture_fn=lambda: self.fx.delete_subscription(subscription_id)) - - def roaming_subscribe(self, phone: str, sink: str, *, types: list[str] | None = None, sink_token: str | None = None, expire: datetime | None = None, initial_event: bool = True) -> NacResult: - p = normalize_phone(phone) - types = types or [ - "org.camaraproject.device-roaming-status-subscriptions.v0.roaming-on", - "org.camaraproject.device-roaming-status-subscriptions.v0.roaming-off", - "org.camaraproject.device-roaming-status-subscriptions.v0.roaming-change-country", - ] - body = self._sub_body(p, sink, types, sink_token, expire, None, initial_event) - return self._call("device-roaming-status-subscriptions", "POST", P["roaming_subs"], json=body, phone=p, fixture_fn=lambda: self.fx.create_subscription("roaming", body)) - - def roaming_unsubscribe(self, subscription_id: str) -> NacResult: - return self._call("device-roaming-status-subscriptions", "DELETE", f"{P['roaming_subs']}/{subscription_id}", fixture_fn=lambda: self.fx.delete_subscription(subscription_id)) - # ---------------------------------------------------------------- QoD def qod_create(self, phone: str, app_server_ipv4: str, qos_profile: str, duration_s: int = 3600, sink: str | None = None) -> NacResult: p = normalize_phone(phone) diff --git a/packages/nac_client/fixtures.py b/packages/nac_client/fixtures.py index ee7aa28..3cb5562 100644 --- a/packages/nac_client/fixtures.py +++ b/packages/nac_client/fixtures.py @@ -3,24 +3,11 @@ Profile (per phone) → produces responses in CAMARA format. Both NacClient's `fixture` mode and the `apps/simulator` HTTP mock server use this class, so a single fixture set behaves the same in both places. -Profile fields (all optional, merged with DEFAULT_PROFILE): - recycled_date "YYYY-MM-DD" | None the number was assigned to a new subscriber on this date - tenure_since "YYYY-MM-DD" date the current subscriber got the line - contract_type PAYG | PAYM | Business - age_check "true" | "false" | "not_available" - verified_status bool whether the operator verified identity with a document - kyc {name, given_name, family_name, birthdate, address, email, id_document} - number_verified bool Number Verification answer - device_phone_number str | None the device's real number (verify: for the equality check) - sim_swap_at ISO datetime | None - device_swap_at ISO datetime | None - call_forwarding ["inactive"] | ["unconditional"] | ["conditional_busy", ...] +Profile fields (all optional, merged with DEFAULT_PROFILE) — only what the six APIs HeatShield calls read: reachable bool ; connectivity ["DATA","SMS"] - roaming bool ; country_code int ; country_name [str] location {"lat":..,"lng":..,"radius":..} | None (None → location UNKNOWN) congestion Low | Medium | High qod_available bool - consent {"": "valid" | "PENDING" | "REVOKED"} (default: all valid) latency_ms int artificial latency (demo: slow API) fail None | "timeout" | 500 | 429 | 403 error injection (demo: circuit breaker) """ @@ -30,33 +17,18 @@ import json import math import uuid -from datetime import datetime, timedelta, timezone, date +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Callable from .errors import NacError, NacTimeout DEFAULT_PROFILE: dict[str, Any] = { - "recycled_date": None, - "tenure_since": "2018-01-01", - "contract_type": "PAYM", - "age_check": "true", - "verified_status": True, - "kyc": {}, - "number_verified": True, - "device_phone_number": None, - "sim_swap_at": None, - "device_swap_at": None, - "call_forwarding": ["inactive"], "reachable": True, "connectivity": ["DATA", "SMS"], - "roaming": False, - "country_code": 90, - "country_name": ["Turkey"], "location": None, "congestion": "Low", "qod_available": True, - "consent": {}, "latency_ms": 0, "fail": None, } @@ -71,12 +43,6 @@ def _parse_dt(v: str | None) -> datetime | None: return dt -def _parse_date(v: str | None) -> date | None: - if not v: - return None - return date.fromisoformat(v[:10]) - - def haversine_m(lat1: float, lng1: float, lat2: float, lng2: float) -> float: r = 6371000.0 p1, p2 = math.radians(lat1), math.radians(lat2) @@ -149,90 +115,6 @@ def preflight(self, api: str, phone: str | None) -> None: raise error_from_status(api, fail, "fixture: injected fault") # ---- CAMARA responses ------------------------------------------------- - def number_recycling(self, phone: str, specified_date: str) -> dict: - prof = self.get(phone) - rd = _parse_date(prof.get("recycled_date")) - sd = _parse_date(specified_date) - return {"phoneNumberRecycled": bool(rd and sd and rd > sd)} - - def call_forwardings(self, phone: str) -> list[str]: - return list(self.get(phone).get("call_forwarding") or ["inactive"]) - - def unconditional_call_forwarding(self, phone: str) -> dict: - return {"active": "unconditional" in self.call_forwardings(phone)} - - def kyc_tenure(self, phone: str, tenure_date: str) -> dict: - prof = self.get(phone) - since = _parse_date(prof.get("tenure_since")) - td = _parse_date(tenure_date) - # tenureDateCheck: has the subscriber been on this line since tenureDate (or earlier)? - return {"tenureDateCheck": bool(since and td and since <= td), "contractType": prof.get("contract_type")} - - def tenure_days(self, phone: str) -> int: - since = _parse_date(self.get(phone).get("tenure_since")) - return (self.now().date() - since).days if since else 0 - - def kyc_age(self, phone: str, age_threshold: int, **fields) -> dict: - prof = self.get(phone) - out = {"ageCheck": prof.get("age_check", "not_available"), "verifiedStatus": prof.get("verified_status")} - if fields: - out["identityMatchScore"] = self._match_score(prof, fields) - return out - - def _match_score(self, prof: dict, fields: dict) -> int: - kyc = prof.get("kyc") or {} - scores = [] - for k, v in fields.items(): - if v is None: - continue - ref = kyc.get(k) - if ref is None: - continue - scores.append(100 if str(ref).strip().lower() == str(v).strip().lower() else _fuzzy(str(ref), str(v))) - return int(sum(scores) / len(scores)) if scores else 0 - - def kyc_match(self, phone: str, **fields) -> dict: - prof = self.get(phone) - kyc = prof.get("kyc") or {} - out: dict[str, Any] = {} - for k, v in fields.items(): - if v is None: - continue - camel = _camel(k) - ref = kyc.get(k) - if ref is None: - out[f"{camel}Match"] = "not_available" - continue - score = 100 if str(ref).strip().lower() == str(v).strip().lower() else _fuzzy(str(ref), str(v)) - out[f"{camel}Match"] = "true" if score >= 80 else "false" - if k not in ("birthdate", "id_document", "gender", "country"): - out[f"{camel}MatchScore"] = score - return out - - def number_verify(self, phone: str) -> dict: - prof = self.get(phone) - dev = prof.get("device_phone_number") - if dev is not None: - return {"devicePhoneNumberVerified": dev == phone} - return {"devicePhoneNumberVerified": bool(prof.get("number_verified"))} - - def sim_swap_check(self, phone: str, max_age_h: int | None) -> dict: - at = _parse_dt(self.get(phone).get("sim_swap_at")) - window = timedelta(hours=max_age_h if max_age_h is not None else 240) - return {"swapped": bool(at and self.now() - at <= window)} - - def sim_swap_date(self, phone: str) -> dict: - at = self.get(phone).get("sim_swap_at") - return {"latestSimChange": at} - - def device_swap_check(self, phone: str, max_age_h: int | None) -> dict: - at = _parse_dt(self.get(phone).get("device_swap_at")) - window = timedelta(hours=max_age_h if max_age_h is not None else 240) - return {"swapped": bool(at and self.now() - at <= window)} - - def device_swap_date(self, phone: str) -> dict: - return {"latestDeviceChange": self.get(phone).get("device_swap_at"), "monitoredPeriod": 120} - def reachability(self, phone: str) -> dict: prof = self.get(phone) return { @@ -242,16 +124,6 @@ def reachability(self, phone: str) -> dict: "connectivity": list(prof.get("connectivity") or []) if prof.get("reachable") else [], } - def roaming(self, phone: str) -> dict: - prof = self.get(phone) - return { - "device": {"phoneNumber": phone}, - "lastStatusTime": self.now().isoformat(), - "roaming": bool(prof.get("roaming")), - "countryCode": prof.get("country_code"), - "countryName": prof.get("country_name"), - } - def location_retrieve(self, phone: str, max_age: int | None = None) -> dict: loc = self.get(phone).get("location") if not loc: @@ -301,20 +173,6 @@ def congestion_query(self, phone: str, start=None, end=None) -> list[dict]: "confidenceLevel": 85, }] - def consent(self, phone: str, scopes: list[str], purpose: str) -> dict: - prof = self.get(phone) - table = prof.get("consent") or {} - info = [] - for s in scopes: - st = table.get(s, "valid") - info.append({ - "scopes": [s], "purpose": purpose, - "statusValidForProcessing": st == "valid", - "statusReason": None if st == "valid" else st, - "expirationDate": None, - }) - return {"statusInfo": info, "captureUrl": None if all(i["statusValidForProcessing"] for i in info) else "https://consent.example/capture"} - # ---- subscriptions / QoD ---------------------------------------------- def create_subscription(self, kind: str, body: dict) -> dict: sid = str(uuid.uuid4()) @@ -392,16 +250,3 @@ def cloud_event(self, event_type: str, subscription_id: str, phone: str, extra: "data": {"subscriptionId": subscription_id, "device": {"phoneNumber": phone}, **(extra or {})}, } - -def _camel(snake: str) -> str: - parts = snake.split("_") - return parts[0] + "".join(p.capitalize() for p in parts[1:]) - - -def _fuzzy(a: str, b: str) -> int: - """Simple similarity: shared-character ratio (0-100). Good enough for the prototype.""" - a, b = a.lower(), b.lower() - if not a or not b: - return 0 - from difflib import SequenceMatcher - return int(SequenceMatcher(None, a, b).ratio() * 100) diff --git a/packages/nac_client/privacy.py b/packages/nac_client/privacy.py index 580a80e..e5de92a 100644 --- a/packages/nac_client/privacy.py +++ b/packages/nac_client/privacy.py @@ -27,7 +27,7 @@ def normalize_phone(phone: str) -> str: def mask_phone(phone: str) -> str: - """+905551234567 -> +90555***4567 . The single format for logs and the UI.""" + """+999991234567 -> +99999***4567 . The single format for logs and the UI.""" try: p = normalize_phone(phone) except ValueError: diff --git a/tests/test_nac_client.py b/tests/test_nac_client.py index 1a3eefa..ea53f1b 100644 --- a/tests/test_nac_client.py +++ b/tests/test_nac_client.py @@ -10,9 +10,9 @@ NOW = datetime(2026, 8, 16, 9, 0, tzinfo=timezone.utc) PROFILES = { - "+905551110001": {"recycled_date": "2026-06-30", "tenure_since": "2026-06-30", "call_forwarding": ["unconditional"], "sim_swap_at": "2026-08-15T20:00:00Z", "age_check": "false", "location": {"lat": 41.0, "lng": 29.0, "radius": 300}}, - "+905551110002": {"fail": 500}, - "+905551110003": {"latency_ms": 50}, + "+999991110001": {"location": {"lat": 41.0, "lng": 29.0, "radius": 300}}, + "+999991110002": {"fail": 500}, + "+999991110003": {"latency_ms": 50}, } @@ -22,53 +22,33 @@ def client(): def test_mask_and_hash(): - assert mask_phone("+905551234567") == "+90555***4567" - assert mask_phone("0555 123 45 67") == "+05551***4567" or mask_phone("+90 555 123 45 67") == "+90555***4567" - assert hash_phone("+905551234567", "s") == hash_phone("+90 555 123 45 67", "s") - assert hash_phone("+905551234567", "s") != hash_phone("+905551234568", "s") + assert mask_phone("+999991234567") == "+99999***4567" + assert mask_phone("+99 999 123 45 67") == "+99999***4567" + assert hash_phone("+999991234567", "s") == hash_phone("+99 999 123 45 67", "s") + assert hash_phone("+999991234567", "s") != hash_phone("+999991234568", "s") with pytest.raises(ValueError): normalize_phone("abc") -def test_number_recycling_relative_to_specified_date(client): - assert client.number_recycling("+905551110001", "2026-01-01").data["phoneNumberRecycled"] is True - assert client.number_recycling("+905551110001", "2026-07-15").data["phoneNumberRecycled"] is False - assert client.number_recycling("+905559999999", "2020-01-01").data["phoneNumberRecycled"] is False - - -def test_call_forwarding_and_sim_swap(client): - assert client.unconditional_call_forwarding("+905551110001").data["active"] is True - assert client.call_forwardings("+905559999999").data == ["inactive"] - assert client.sim_swap_check("+905551110001", 72).data["swapped"] is True - assert client.sim_swap_check("+905551110001", 1).data["swapped"] is False - - -def test_kyc(client): - assert client.kyc_tenure("+905551110001", "2026-01-01").data["tenureDateCheck"] is False # line was acquired on 30 June, not since January - assert client.kyc_tenure("+905551110001", "2026-08-01").data["tenureDateCheck"] is True - assert client.kyc_age("+905551110001", 18).data["ageCheck"] == "false" - assert client.kyc_age("+905559999999", 18).data["ageCheck"] == "true" - - def test_location_verify(client): - assert client.location_verify("+905551110001", 41.0, 29.0, 1000).data["verificationResult"] == "TRUE" - assert client.location_verify("+905551110001", 41.05, 29.0, 1000).data["verificationResult"] == "FALSE" - assert client.location_verify("+905559999999", 41.0, 29.0, 1000).data["verificationResult"] == "UNKNOWN" + assert client.location_verify("+999991110001", 41.0, 29.0, 1000).data["verificationResult"] == "TRUE" + assert client.location_verify("+999991110001", 41.05, 29.0, 1000).data["verificationResult"] == "FALSE" + assert client.location_verify("+999999999999", 41.0, 29.0, 1000).data["verificationResult"] == "UNKNOWN" def test_result_masks_phone_in_request_summary(client): - r = client.reachability("+905551110001") - assert r.request["device"]["phoneNumber"] == "+90555***0001" + r = client.reachability("+999991110001") + assert r.request["device"]["phoneNumber"] == "+99999***0001" assert r.source == "fixture" and r.latency_ms >= 0 def test_retry_then_circuit_opens(client): with pytest.raises(NacError) as e: - client.reachability("+905551110002") + client.reachability("+999991110002") assert e.value.kind == "server" # breaker_threshold=2 → first call 1+1 retry = 2 failures → circuit open with pytest.raises(NacCircuitOpen): - client.reachability("+905551110002") + client.reachability("+999991110002") assert client.health()["breaker"]["device-reachability-status"]["open"] is True @@ -77,16 +57,21 @@ def test_simulator_http_roundtrip(): import simulator.main as sim sim.fx = FixtureBackend(PROFILES, now=lambda: NOW) http = TestClient(sim.app, base_url="http://sim") - c = NacClient(NacConfig(mode="simulator", base_url="http://sim", rapidapi_key="k", oauth_token="t"), http=http) - assert c.number_recycling("+905551110001", "2026-01-01").data["phoneNumberRecycled"] is True - assert c.unconditional_call_forwarding("+905551110001").data["active"] is True - assert c.reachability("+905551110001").data["reachable"] is True - sub = c.geofence_subscribe("+905551110001", 41.0, 29.0, 500, "http://sink.local/webhook", sink_token="abc").data + c = NacClient(NacConfig(mode="simulator", base_url="http://sim", rapidapi_key="k"), http=http) + # the six APIs the product calls, over real HTTP on the real CAMARA paths + assert c.reachability("+999991110001").data["reachable"] is True + assert c.location_verify("+999991110001", 41.0, 29.0, 1000, 600).data["verificationResult"] == "TRUE" + assert c.location_retrieve("+999991110001", 60).data["area"]["center"]["latitude"] == 41.0 + assert isinstance(c.congestion_query("+999991110001").data, list) + qod = c.qod_create("+999991110001", "203.0.113.10", "QOS_E", 600).data + assert qod["sessionId"] and c.qod_get(qod["sessionId"]).data["sessionId"] == qod["sessionId"] + c.qod_delete(qod["sessionId"]) + sub = c.geofence_subscribe("+999991110001", 41.0, 29.0, 500, "http://sink.local/webhook", sink_token="abc").data assert sub["status"] == "ACTIVE" and sub["id"] assert c.geofence_get(sub["id"]).data["id"] == sub["id"] assert c.geofence_delete(sub["id"]).data["id"] == sub["id"] with pytest.raises(NacError) as e: - c.reachability("+905551110002") + c.reachability("+999991110002") assert e.value.kind == "server" diff --git a/tests/test_verify_live.py b/tests/test_verify_live.py new file mode 100644 index 0000000..4efed8d --- /dev/null +++ b/tests/test_verify_live.py @@ -0,0 +1,51 @@ +"""The live probe's logic, exercised offline against the fixture backend. + +Only the probe functions are called — never `main()`, which would read `.env`. No network is touched. +""" +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools")) +import verify_live as VL # noqa: E402 + +from nac_client import FixtureBackend, NacClient, NacConfig # noqa: E402 + +PHONE = VL.DEFAULT_PROBE_PHONE + + +@pytest.fixture +def client(): + fx = FixtureBackend({PHONE: {"location": {"lat": 25.2854, "lng": 51.5310, "radius": 300}}}) + return NacClient(NacConfig(mode="fixture"), fixtures=fx) + + +def test_the_probe_covers_exactly_the_apis_the_product_calls(): + apis = {p.paths[0].split("/")[0] if not p.paths[0].startswith("device-status") else p.paths[0].split("/")[1] + for p in VL.PROBES} + assert apis == {"device-reachability-status", "location-retrieval", "location-verification", + "congestion-insights", "geofencing-subscriptions", "quality-on-demand"} + assert VL.DEFAULT_PROBE_PHONE.startswith("+99999") # a sandbox test device, never a real-format number + + +def test_every_probe_succeeds_and_everything_created_is_deleted(client, tmp_path): + ctx = VL.ProbeCtx(phone=PHONE, sink="http://127.0.0.1:8000/webhooks/probe") + results = VL.run_probes(client, ctx, include_write=True, only=None) + assert [r["key"] for r in results if not r.get("ok")] == [] + assert {k for k, _ in ctx.created} == {"geofencing", "qod"} + cleaned = VL.cleanup(client, ctx) + assert cleaned and all(c["deleted"] for c in cleaned) + + meta = {"started": "2026-09-27T12:00:00+00:00", "mode": "fixture", "base_url": "-", "phone_masked": "x", + "sink": ctx.sink, "has_key": False, "include_write": True} + j, md = VL.write_report(results, meta, cleaned, out_dir=tmp_path) + assert j.exists() and md.exists() + assert PHONE not in j.read_text(encoding="utf-8") and PHONE not in md.read_text(encoding="utf-8") + + +def test_write_probes_are_skipped_without_include_write(client): + ctx = VL.ProbeCtx(phone=PHONE, sink="http://127.0.0.1:8000/webhooks/probe") + results = VL.run_probes(client, ctx, include_write=False, only=None) + assert {r["key"] for r in results if r.get("skipped")} == {"geofencing_subs", "qod_sessions"} + assert ctx.created == [] diff --git a/tools/verify_live.py b/tools/verify_live.py index 32d5ec4..0a298c7 100644 --- a/tools/verify_live.py +++ b/tools/verify_live.py @@ -1,31 +1,29 @@ -"""verify_live.py — Nokia NaC live availability probe (spec rule 0.3). +"""verify_live.py — live availability probe for the six CAMARA APIs HeatShield calls. -Tries ALL CAMARA endpoints the portfolio uses one by one, records the result and -automatically fills the "SIMULATOR planında test" column in each project's -`docs/api-availability.md` table. Goal: reduce manual table filling to zero. +Calls each endpoint once, records status, latency and the response shape, cleans up anything it +created, and writes a masked report to `evidence/probe-.json` and `.md`. + +The 9 Sep 2026 probe in `evidence/` was run with the version of this tool at tag `v0.1-hackathon`. +That version also probed APIs the product never calls (KYC, SIM swap, number verification, …); +they have since been removed from the client, so this tool no longer probes them. Usage ----- - # 1) First, a rehearsal against the local simulator (no key needed, breaks nothing) + # rehearsal against the local simulator (no key, no network beyond localhost) python tools/verify_live.py --mode simulator --base http://127.0.0.1:8081 - # 2) After the Nokia account is opened: real calls + update the tables - python tools/verify_live.py --mode live --patch + # real calls against Nokia Network as Code — spends quota + python tools/verify_live.py --mode live - # 3) Also try endpoints that CREATE RECORDS, such as subscriptions / QoD (creates and deletes) - python tools/verify_live.py --mode live --include-write --sink https:///webhooks/probe --patch + # also the endpoints that CREATE records (geofence subscription, QoD session); both are deleted again + python tools/verify_live.py --mode live --include-write --sink https:///webhooks/probe -Environment variables (.env or shell): - NAC_RAPIDAPI_KEY RapidAPI key (required, live) - NAC_OAUTH_TOKEN Bearer token for passthrough/ endpoints - NAC_OAUTH_TOKEN_URL if given, a token is obtained via client-credentials - NAC_OAUTH_CLIENT_ID ^ - NAC_OAUTH_CLIENT_SECRET ^ - NAC_PROBE_PHONE probe number (default +905551234567) - PUBLIC_BASE_URL public HTTPS base address for the subscription sink +Environment (.env or shell): + NAC_RAPIDAPI_KEY RapidAPI key (required for live) + NAC_PROBE_PHONE probe number (default +99999000001000, a Nokia sandbox test device) + PUBLIC_BASE_URL public HTTPS base for the subscription sink -Rule: the raw number appears in no output (`_mask_deep`). No fabricated results are written — -an endpoint that was not tried stays "☐ yapılmadı" (not done) in the table. +The raw number appears in no output (`_mask_deep`), and nothing is written for an endpoint that was not tried. """ from __future__ import annotations @@ -35,32 +33,25 @@ import re import sys from dataclasses import dataclass, field -from datetime import datetime, timedelta, timezone +from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable -SHARED = Path(__file__).resolve().parent.parent # repository root (heatshield/) -ROOT = SHARED -sys.path.insert(0, str(SHARED)) # packages/ lives at the repository root - -from packages.nac_client import NacClient, NacConfig, NacError, mask_phone # noqa: E402 -from packages.nac_client.client import P # verified path dictionary # noqa: E402 +from _common import EVIDENCE, ROOT, load_env # noqa: E402 (repo root + package path) +from nac_client import NacClient, NacConfig, NacError, mask_phone # noqa: E402 +from nac_client.client import P # the verified path table # noqa: E402 -# Projects to patch. `_template` is DELIBERATELY left out: its table is the empty skeleton copied -# into new projects ("the project agent fills it in") — it is not polluted with probe results. -PROJECTS = ["heatshield"] - -# ---------------------------------------------------------------- probe definitions +DEFAULT_PROBE_PHONE = "+99999000001000" @dataclass class Probe: key: str - label: str # name shown in the table/report - paths: list[str] # path fragments used to find this row in the table + label: str + paths: list[str] run: Callable[[NacClient, "ProbeCtx"], Any] - write: bool = False # does it create a record (subscription/QoD) → requires --include-write - note: str = "" # known limitation in live use + write: bool = False # creates a record (subscription / QoD session) → needs --include-write + note: str = "" # a known limitation in live use @dataclass @@ -73,12 +64,8 @@ class ProbeCtx: created: list[tuple[str, str]] = field(default_factory=list) # (kind, id) → cleanup -def _days_ago(n: int) -> str: - return (datetime.now(timezone.utc) - timedelta(days=n)).date().isoformat() - - -def _cleanup_sub(ctx: ProbeCtx, kind: str, res: Any) -> Any: - """Take the id from the subscription/session response and add it to the cleanup list.""" +def _track(ctx: ProbeCtx, kind: str, res: Any) -> Any: + """Take the id from a subscription/session response and add it to the cleanup list.""" sid = (res.data or {}).get("subscriptionId") or (res.data or {}).get("sessionId") or (res.data or {}).get("id") if sid: ctx.created.append((kind, str(sid))) @@ -86,75 +73,36 @@ def _cleanup_sub(ctx: ProbeCtx, kind: str, res: Any) -> Any: PROBES: list[Probe] = [ - # --- passthrough (Bearer + RapidAPI key) ----------------------------------- - Probe("number_recycling", "Number Recycling", [P["number_recycling"]], - lambda c, x: c.number_recycling(x.phone, _days_ago(30))), - Probe("call_forwardings", "Call Forwarding Signal — tümü", [P["call_forwardings"]], - lambda c, x: c.call_forwardings(x.phone)), - Probe("unconditional_cf", "Call Forwarding Signal — koşulsuz", [P["unconditional_cf"]], - lambda c, x: c.unconditional_call_forwarding(x.phone)), - Probe("kyc_tenure", "KYC Tenure", [P["kyc_tenure"]], - lambda c, x: c.kyc_tenure(x.phone, _days_ago(90))), - Probe("kyc_age", "KYC Age Verification", [P["kyc_age"]], - lambda c, x: c.kyc_age(x.phone, 18)), - Probe("kyc_match", "KYC Match", [P["kyc_match"]], - lambda c, x: c.kyc_match(x.phone, name="Test Kullanici")), - Probe("number_verify", "Number Verification", [P["number_verify"]], - lambda c, x: c.number_verify(x.phone), - note="Canlıda 3-legged OIDC + mobil veri ister; laboratuvar Wi‑Fi'ından 401/403 beklenir."), - Probe("sim_swap_check", "SIM Swap check", [P["sim_swap_check"]], - lambda c, x: c.sim_swap_check(x.phone, 72)), - Probe("sim_swap_date", "SIM Swap date", [P["sim_swap_date"]], - lambda c, x: c.sim_swap_date(x.phone)), - Probe("device_swap_check", "Device Swap check", [P["device_swap_check"]], - lambda c, x: c.device_swap_check(x.phone, 24)), - Probe("device_swap_date", "Device Swap date", [P["device_swap_date"]], - lambda c, x: c.device_swap_date(x.phone)), - Probe("consent", "Consent Info", [P["consent"]], - lambda c, x: c.consent(x.phone, ["device-swap:read", "sim-swap:read"], "identity-assurance")), - # --- device-status / location (RapidAPI key only) -------------------------- Probe("reachability", "Device Reachability Status", [P["reachability"]], lambda c, x: c.reachability(x.phone)), - Probe("roaming", "Device Roaming Status", [P["roaming"]], - lambda c, x: c.roaming(x.phone)), Probe("location_retrieve", "Location Retrieval", [P["location_retrieve"]], lambda c, x: c.location_retrieve(x.phone, 3600)), Probe("location_verify", "Location Verification", [P["location_verify"]], lambda c, x: c.location_verify(x.phone, x.lat, x.lng, x.radius_m, 3600)), Probe("congestion_query", "Congestion Insights", [P["congestion_query"]], lambda c, x: c.congestion_query(x.phone)), - # --- subscriptions and QoD: CREATE RECORDS ---------------------------------- Probe("geofencing_subs", "Geofencing Subscriptions", [P["geofencing_subs"]], - lambda c, x: _cleanup_sub(x, "geofencing", c.geofence_subscribe(x.phone, x.lat, x.lng, x.radius_m, x.sink)), - write=True, note="Canlıda sink public HTTPS olmalı (ngrok/cloudflare tüneli)."), - Probe("reachability_subs", "Device Reachability Status Subscriptions", [P["reachability_subs"]], - lambda c, x: _cleanup_sub(x, "reachability", c.reachability_subscribe(x.phone, x.sink)), - write=True, note="Canlıda sink public HTTPS olmalı."), - Probe("roaming_subs", "Device Roaming Status Subscriptions", [P["roaming_subs"]], - lambda c, x: _cleanup_sub(x, "roaming", c.roaming_subscribe(x.phone, x.sink)), - write=True, note="Canlıda sink public HTTPS olmalı."), + lambda c, x: _track(x, "geofencing", c.geofence_subscribe(x.phone, x.lat, x.lng, x.radius_m, x.sink)), + write=True, note="Live: the sink must be public HTTPS (an ngrok/cloudflare tunnel)."), Probe("qod_sessions", "Quality on Demand", [P["qod_sessions"]], - lambda c, x: _cleanup_sub(x, "qod", c.qod_create(x.phone, "203.0.113.10", "QOS_E", 600)), - write=True, note="qosProfile adı operatöre göre değişir; 400 alırsan portaldaki profil listesine bak."), - Probe("qod_retrieve", "Quality on Demand — oturum listesi", [P["qod_retrieve"]], + lambda c, x: _track(x, "qod", c.qod_create(x.phone, "203.0.113.10", "QOS_E", 600)), + write=True, note="qosProfile names vary by operator; on a 400, check the profile list in the portal."), + Probe("qod_retrieve", "Quality on Demand — session list", [P["qod_retrieve"]], lambda c, x: c.qod_list(x.phone)), ] -# ---------------------------------------------------------------- helpers - _PHONE_RE = re.compile(r"\+[0-9]{8,15}") def _shape(data: Any) -> list[str]: - """The shape of the response — "which fields came back", for the jury/table. - On some endpoints CAMARA returns an array (e.g. call-forwardings); then there are no field names.""" + """Which fields came back. Some CAMARA endpoints return an array; then there are no field names.""" if isinstance(data, dict): return sorted(data.keys()) if isinstance(data, list): inner = sorted({k for it in data if isinstance(it, dict) for k in it}) - return [f"dizi[{len(data)}]"] + inner + return [f"array[{len(data)}]"] + inner if data is None or data == {}: - return ["(boş gövde)"] + return ["(empty body)"] return [type(data).__name__] @@ -168,64 +116,22 @@ def _mask_deep(value: Any) -> Any: return value -def load_dotenv(path: Path) -> int: - """Simple .env reader (without adding a python-dotenv dependency).""" - if not path.exists(): - return 0 - n = 0 - for line in path.read_text(encoding="utf-8-sig").splitlines(): - line = line.strip() - if not line or line.startswith("#") or "=" not in line: - continue - k, v = line.split("=", 1) - k, v = k.strip(), v.strip().strip('"').strip("'") - if k and k not in os.environ: - os.environ[k] = v - n += 1 - return n - - -def fetch_oauth_token() -> str | None: - """If NAC_OAUTH_TOKEN_URL is given, obtains a token via client-credentials.""" - url = os.environ.get("NAC_OAUTH_TOKEN_URL") - cid = os.environ.get("NAC_OAUTH_CLIENT_ID") - secret = os.environ.get("NAC_OAUTH_CLIENT_SECRET") - if not (url and cid and secret): - return None - import httpx - - scope = os.environ.get("NAC_OAUTH_SCOPE") - data = {"grant_type": "client_credentials", **({"scope": scope} if scope else {})} - r = httpx.post(url, data=data, auth=(cid, secret), timeout=15.0) - r.raise_for_status() - tok = r.json().get("access_token") - if not tok: - raise RuntimeError(f"token yanıtında access_token yok: {list(r.json())}") - return tok - - -# ---------------------------------------------------------------- probe runner - -VERDICT_OK = "✅" -VERDICT_REACHED = "◐" - -# CAMARA error bodies carry codes (e.g. LOCATION_RETRIEVAL.UNABLE_TO_LOCATE). If we saw such a code, -# the endpoint REALLY EXISTS and speaks CAMARA — the test data just did not produce a positive result. -# DOMAIN.CODE — the most informative. At least 2 characters on each side: also catches short domain -# names such as `QOD.SESSION_NOT_FOUND`, but not "A.B" initials in plain text. +# CAMARA error bodies carry codes (e.g. LOCATION_RETRIEVAL.UNABLE_TO_LOCATE). If we saw such a code, the +# endpoint REALLY EXISTS and speaks CAMARA — the test data just did not produce a positive result. +# DOMAIN.CODE is the most informative; at least 2 characters on each side, so short domains such as +# `QOD.SESSION_NOT_FOUND` match but "A.B" initials in plain text do not. _DOTTED_RE = re.compile(r"\b([A-Z][A-Z0-9_]+\.[A-Z][A-Z0-9_]+)\b") _BARE_RE = re.compile(r"\b([A-Z][A-Z0-9]{2,}(?:_[A-Z0-9]+)+)\b") # BARE_CODE def camara_code(rec: dict) -> str | None: - """Extracts the CAMARA error code from the error body. The `DOMAIN.CODE` form is preferred over `CODE`.""" + """The CAMARA error code in the error body; `DOMAIN.CODE` is preferred over `CODE`.""" if rec.get("ok") or rec.get("skipped"): return None raw = str(rec.get("message") or "") texts = [raw] - # if the body is JSON, also scan the inner `code`/`message` fields i, j = raw.find("{"), raw.rfind("}") - if 0 <= i < j: + if 0 <= i < j: # a JSON body: also scan its `code`/`message` try: obj = json.loads(raw[i:j + 1]) if isinstance(obj, dict): @@ -242,7 +148,7 @@ def camara_code(rec: dict) -> str | None: def endpoint_reached(rec: dict) -> bool: """Does the endpoint exist? 200 or a CAMARA-coded business-rule error → yes. - 401/403 (authorization), timeout/network/circuit breaker → no, no information.""" + 401/403, timeout, network or an open circuit → no information.""" if rec.get("ok"): return True if rec.get("kind") in ("auth", "timeout", "network", "circuit_open", "probe_error"): @@ -250,29 +156,6 @@ def endpoint_reached(rec: dict) -> bool: return camara_code(rec) is not None -def verdict_cell(rec: dict, stamp: str) -> str: - """One-line verdict to write into the last column of api-availability.md.""" - if rec["ok"]: - return f"{VERDICT_OK} 200 · {rec['latency_ms']} ms · {stamp}" - kind, status = rec.get("kind"), rec.get("status") - code = camara_code(rec) - if code: - # the endpoint is up and returned a business-rule error — it must not look "missing" in the table - return f"{VERDICT_REACHED} uç var · {status or ''} `{code}` · {stamp}".replace("· ·", "·") - label = { - "auth": "plan/yetki kapsamı dışı", - "not_found": "path bulunamadı", - "bad_request": "gövde reddedildi", - "rate_limit": "kota doldu", - "server": "sunucu hatası", - "timeout": "zaman aşımı", - "network": "ağ hatası", - "circuit_open": "devre kesici açık", - }.get(kind or "", kind or "hata") - st = f" {status}" if status else "" - return f"⚠{st} {label} · {stamp}" - - def run_probes(client: NacClient, ctx: ProbeCtx, include_write: bool, only: set[str] | None) -> list[dict]: out: list[dict] = [] for pr in PROBES: @@ -280,8 +163,8 @@ def run_probes(client: NacClient, ctx: ProbeCtx, include_write: bool, only: set[ continue if pr.write and not include_write: out.append({"key": pr.key, "label": pr.label, "paths": pr.paths, "skipped": True, - "reason": "kayıt oluşturan uç — --include-write ile denenir", "note": pr.note}) - print(f" ⏭ {pr.label:<44} atlandı (--include-write gerekli)") + "reason": "creates a record — tried only with --include-write", "note": pr.note}) + print(f" ⏭ {pr.label:<40} skipped (needs --include-write)") continue try: res = pr.run(client, ctx) @@ -289,204 +172,46 @@ def run_probes(client: NacClient, ctx: ProbeCtx, include_write: bool, only: set[ "latency_ms": res.latency_ms, "source": res.source, "correlator": res.correlator, "request": _mask_deep(res.request), "response_keys": _shape(res.data), "response": _mask_deep(res.data), "note": pr.note} - print(f" ✅ {pr.label:<44} 200 {res.latency_ms:>5} ms {res.source}") + print(f" ✅ {pr.label:<40} 200 {res.latency_ms:>5} ms {res.source}") except NacError as e: rec = {"key": pr.key, "label": pr.label, "paths": pr.paths, "ok": False, **e.to_dict(), "note": pr.note} - code = camara_code(rec) - rec["camara_code"] = code + rec["camara_code"] = camara_code(rec) rec["endpoint_reached"] = endpoint_reached(rec) - if code: - print(f" ◐ {pr.label:<44} uç var, iş kuralı hatası: {e.status} {code}") + if rec["camara_code"]: + print(f" ◐ {pr.label:<40} endpoint exists, business-rule error: {e.status} {rec['camara_code']}") else: - print(f" ⚠ {pr.label:<44} {e.kind}{f' {e.status}' if e.status else ''}") - except Exception as e: # do not let the probe crash + print(f" ⚠ {pr.label:<40} {e.kind}{f' {e.status}' if e.status else ''}") + except Exception as e: # noqa: BLE001 — one broken probe must not stop the others rec = {"key": pr.key, "label": pr.label, "paths": pr.paths, "ok": False, "kind": "probe_error", "message": f"{type(e).__name__}: {e}", "note": pr.note} - print(f" ✖ {pr.label:<44} sonda hatası: {type(e).__name__}") + print(f" ✖ {pr.label:<40} probe error: {type(e).__name__}") out.append(rec) return out def cleanup(client: NacClient, ctx: ProbeCtx) -> list[dict]: - """Delete the subscriptions/sessions created during the probe — leave no garbage behind.""" + """Delete the subscriptions/sessions the probe created — leave nothing behind on the platform.""" done = [] for kind, sid in ctx.created: try: if kind == "geofencing": client.geofence_delete(sid) - elif kind == "reachability": - client.reachability_unsubscribe(sid) - elif kind == "roaming": - client.roaming_unsubscribe(sid) elif kind == "qod": client.qod_delete(sid) done.append({"kind": kind, "id": sid, "deleted": True}) - print(f" 🧹 {kind} {sid} silindi") + print(f" 🧹 {kind} {sid} deleted") except NacError as e: done.append({"kind": kind, "id": sid, "deleted": False, "error": e.kind}) - print(f" ⚠ {kind} {sid} silinemedi: {e.kind} — portaldan elle kontrol et") + print(f" ⚠ {kind} {sid} could not be deleted: {e.kind} — check the portal by hand") return done -# ---------------------------------------------------------------- table patcher - -TEST_COL_HINT = "SIMULATOR planında test" -PLACEHOLDER_RE = re.compile(r"^\s*(☐|\[ \]|—|-)?\s*(yapılmadı|yapilmadi)?\s*$", re.IGNORECASE) - - -def _split_row(line: str) -> list[str]: - """Splits a Markdown table row into cells (excluding the leading/trailing pipe).""" - s = line.strip() - if s.startswith("|"): - s = s[1:] - if s.endswith("|"): - s = s[:-1] - return s.split("|") - - -def _join_row(cells: list[str]) -> str: - return "| " + " | ".join(c.strip() for c in cells) + " |" - - -def patch_availability(project_dir: Path, results: list[dict], stamp: str, dry: bool) -> dict: - """Fills the test column in `docs/api-availability.md` by path match.""" - doc = project_dir / "docs" / "api-availability.md" - if not doc.exists(): - return {"project": project_dir.name, "error": "api-availability.md yok"} - - lines = doc.read_text(encoding="utf-8").splitlines() - by_path: list[tuple[str, dict]] = [] - for rec in results: - for p in rec["paths"]: - by_path.append((p.strip("/"), rec)) - # longer paths first → so "sim-swap/v0/check" and "sim-swap/v0" do not get mixed up - by_path.sort(key=lambda t: -len(t[0])) - - col_idx: int | None = None - touched: list[str] = [] - unmatched: list[str] = [] - skipped_rows: list[str] = [] - - for i, line in enumerate(lines): - if not line.lstrip().startswith("|"): - continue - cells = _split_row(line) - if col_idx is None: - for j, c in enumerate(cells): - if TEST_COL_HINT in c: - col_idx = j - break - continue # we do not patch the header row - if set(line.replace("|", "").replace("-", "").replace(":", "").strip()) == set(): - continue # separator row - if col_idx >= len(cells): - continue - row_text = line - hit = next((rec for p, rec in by_path if p in row_text), None) - if hit is None: - first = cells[0].strip().strip("*")[:40] - if first: - unmatched.append(first) - continue - if hit.get("skipped"): - # not tried → let it STAY "☐ yapılmadı" in the table; we do not write a fabricated verdict - skipped_rows.append(hit["label"]) - continue - cur = cells[col_idx].strip() - if cur.startswith((VERDICT_OK, VERDICT_REACHED, "⚠")): - pass # previous probe result — overwrite it (the most recent wins) - elif not PLACEHOLDER_RE.match(cur): - # there is a hand-written note: add the verdict, keep the note - cells[col_idx] = f"{verdict_cell(hit, stamp)} — {cur}" - lines[i] = _join_row(cells) - touched.append(hit["label"]) - continue - cells[col_idx] = verdict_cell(hit, stamp) - lines[i] = _join_row(cells) - touched.append(hit["label"]) - - # "Canlı test kontrol listesi" (live test checklist) — tick the verified items - any_ok = any(r.get("ok") for r in results) - nv_ok = any(r.get("ok") for r in results if r["key"] == "number_verify") - for i, line in enumerate(lines): - if not line.lstrip().startswith("- [ ]"): - continue - low = line.lower() - if any_ok and ("sign-up" in low or "kayıt" in low or "kayit" in low): - lines[i] = line.replace("- [ ]", "- [x]", 1) + f" _({stamp})_" - elif nv_ok and "number verification" in low: - lines[i] = line.replace("- [ ]", "- [x]", 1) + f" _({stamp})_" - elif any_ok and ("bu tabloya işlendi" in low or "tabloya işlendi" in low or "sonuçlar bu tabloya" in low): - lines[i] = line.replace("- [ ]", "- [x]", 1) + f" _(tools/verify_live.py, {stamp})_" - - new = "\n".join(lines) + "\n" - if not dry: - doc.write_text(new, encoding="utf-8") - return {"project": project_dir.name, "col": col_idx, "patched": touched, - "unmatched": unmatched, "skipped_rows": skipped_rows, "dry": dry} - - -# ---------------------------------------------------------------- master inventory table - -def which_projects(paths: list[str]) -> list[str]: - """Which projects' api-availability.md mention this endpoint — derived; we do not keep a manual list.""" - hits = [] - for name in PROJECTS: - doc = ROOT / name / "docs" / "api-availability.md" - if not doc.exists(): - continue - text = doc.read_text(encoding="utf-8") - if any(p.strip("/") in text for p in paths): - hits.append(name) - return hits - - -def write_master_table(results: list[dict], meta: dict, stamp: str, dry: bool) -> Path: - """A single table of ALL endpoints in the portfolio — under the root docs/.""" - doc = SHARED / "docs" / "api-availability-master.md" - rows = [] - for r in results: - if r.get("skipped"): - cell = "☐ atlandı (`--include-write`)" - else: - cell = verdict_cell(r, stamp) - users = which_projects(r["paths"]) or ["—"] - pt = "Bearer + key" if r["paths"][0].startswith("passthrough/") else "yalnız key" - rows.append(f"| **{r['label']}** | `{r['paths'][0]}` | {pt} | {', '.join(users)} | {cell} |") - - ok = sum(1 for r in results if r.get("ok")) - reached = sum(1 for r in results if not r.get("ok") and not r.get("skipped") and endpoint_reached(r)) - tried = sum(1 for r in results if not r.get("skipped")) - body = ( - "# API Erişilebilirlik — Ana Envanter\n\n" - "> Portföyün kullandığı **bütün** CAMARA uçları tek tabloda. Bu dosyayı elle düzenlemeyin:\n" - "> `python tools/verify_live.py --mode live --patch` her çalıştığında yeniden üretilir.\n" - f"> Son sonda: **{meta['started'][:16]}** · mod `{meta['mode']}` · taban `{meta['base_url']}`" - f" · numara `{meta['phone_masked']}`\n" - f"> Sonuç: **{ok} çalıştı**, {reached} uç var (iş kuralı hatası), {tried - ok - reached} erişilemedi" - f" — denenen {tried}/{len(results)}\n\n" - "Path kaynağı: `network-as-code` SDK v10.0.0 → [`nac-api-reference.md`](nac-api-reference.md).\n" - "Hüküm işaretleri: ✅ 200 döndü · ◐ uç mevcut, CAMARA iş kuralı hatası (401/403 değil) · ⚠ erişilemedi · ☐ denenmedi.\n\n" - "| API | Path | Auth | Kullanan projeler | Son sonda |\n|---|---|---|---|---|\n" - + "\n".join(rows) + "\n\n" - "## Notlar\n" - + "\n".join(f"- **{r['label']}**: {r['note']}" for r in results if r.get("note")) + "\n\n" - "Ayrıntılı çıktı (istek/yanıt gövdeleri, maskeli): [`live-probe/`](live-probe/).\n" - ) - if not dry: - doc.parent.mkdir(parents=True, exist_ok=True) - doc.write_text(body, encoding="utf-8") - return doc - - -# ---------------------------------------------------------------- report - -def write_report(results: list[dict], meta: dict, cleaned: list[dict]) -> tuple[Path, Path]: - out_dir = SHARED / "evidence" +def write_report(results: list[dict], meta: dict, cleaned: list[dict], out_dir: Path = EVIDENCE) -> tuple[Path, Path]: out_dir.mkdir(parents=True, exist_ok=True) slug = meta["started"].replace(":", "").replace("-", "").replace("T", "-")[:15] j = out_dir / f"probe-{slug}.json" - j.write_text(json.dumps({"meta": meta, "results": results, "cleanup": cleaned}, ensure_ascii=False, indent=2), encoding="utf-8") + j.write_text(json.dumps({"meta": meta, "results": results, "cleanup": cleaned}, ensure_ascii=False, indent=2), + encoding="utf-8") ok = [r for r in results if r.get("ok")] reached = [r for r in results if not r.get("ok") and not r.get("skipped") and endpoint_reached(r)] @@ -495,130 +220,89 @@ def write_report(results: list[dict], meta: dict, cleaned: list[dict]) -> tuple[ rows = [] for r in results: if r.get("skipped"): - state = "⏭ atlandı" + state = "⏭ skipped" elif r.get("ok"): state = f"✅ 200 · {r['latency_ms']} ms" elif endpoint_reached(r): - state = f"◐ uç var · {r.get('status','')} `{camara_code(r)}`" + state = f"◐ endpoint exists · {r.get('status', '')} `{camara_code(r)}`" else: state = f"⚠ {r.get('kind')}" + (f" {r['status']}" if r.get("status") else "") keys = ", ".join(r.get("response_keys") or []) if r.get("ok") else (r.get("message", "") or "")[:80] - rows.append(f"| {r['label']} | `{r['paths'][0]}` | {state} | {keys} | {r.get('note','')} |") + rows.append(f"| {r['label']} | `{r['paths'][0]}` | {state} | {keys} | {r.get('note', '')} |") md = out_dir / f"probe-{slug}.md" md.write_text( - f"# NaC canlı sonda raporu — {meta['started'][:16]}\n\n" - f"- Mod: **{meta['mode']}** · taban: `{meta['base_url']}`\n" - f"- Sonda numarası: `{meta['phone_masked']}`\n" - f"- RapidAPI anahtarı: {'var' if meta['has_key'] else 'YOK'} · Bearer token: {'var' if meta['has_token'] else 'YOK'}\n" - f"- Sonuç: **{len(ok)} çalıştı**, {len(reached)} uç var (iş kuralı hatası), {len(bad)} erişilemedi, {len(skipped)} atlandı" - f" — toplam {len(results)}\n" - f"- **Uç mevcut sayılan: {len(ok) + len(reached)}/{len(results) - len(skipped)}**" - " (200 ya da CAMARA kodlu hata; 401/403/timeout hariç)\n\n" - "| API | Path | Sonuç | Yanıt alanları / hata | Not |\n|---|---|---|---|---|\n" + f"# NaC live probe report — {meta['started'][:16]}\n\n" + f"- Mode: **{meta['mode']}** · base: `{meta['base_url']}`\n" + f"- Probe number: `{meta['phone_masked']}`\n" + f"- RapidAPI key: {'present' if meta['has_key'] else 'MISSING'}\n" + f"- Result: **{len(ok)} succeeded**, {len(reached)} endpoints exist (business-rule error), " + f"{len(bad)} unreachable, {len(skipped)} skipped — {len(results)} in total\n\n" + "| API | Path | Result | Response fields / error | Note |\n|---|---|---|---|---|\n" + "\n".join(rows) + "\n\n" - + ("## Temizlik\n" + "\n".join(f"- {c['kind']} `{c['id']}` → {'silindi' if c['deleted'] else 'SİLİNEMEDİ (' + c.get('error','?') + ')'}" for c in cleaned) + "\n" if cleaned else "") - + "\n> Ham telefon numarası bu raporda yoktur (`_mask_deep`).\n", + + ("## Cleanup\n" + "\n".join( + f"- {c['kind']} `{c['id']}` → {'deleted' if c['deleted'] else 'NOT DELETED (' + c.get('error', '?') + ')'}" + for c in cleaned) + "\n" if cleaned else "") + + "\n> No raw phone number appears in this report (`_mask_deep`).\n", encoding="utf-8", ) return j, md -# ---------------------------------------------------------------- main - def main() -> int: - ap = argparse.ArgumentParser(description="Nokia NaC canlı erişilebilirlik sondası") + ap = argparse.ArgumentParser(description="Live availability probe for the six CAMARA APIs HeatShield calls") ap.add_argument("--mode", required=True, choices=["live", "simulator", "fixture"], - help="zorunlu: live gerçek anahtarla gerçek çağrı yapar ve kota harcar") - ap.add_argument("--base", default=None, help="taban URL (varsayılan: moda göre)") - ap.add_argument("--phone", default=None, help="sonda numarası (E.164)") - ap.add_argument("--sink", default=None, help="abonelik webhook adresi (public HTTPS)") - ap.add_argument("--include-write", action="store_true", help="abonelik/QoD gibi kayıt oluşturan uçları da dene") - ap.add_argument("--only", default=None, help="virgülle ayrılmış sonda anahtarları (ör. number_verify,consent)") - ap.add_argument("--patch", action="store_true", help="api-availability.md tablolarını güncelle") - ap.add_argument("--dry-run", action="store_true", help="--patch ile: dosyaya yazma, ne yapacağını göster") - ap.add_argument("--force", action="store_true", - help="canlı olmayan sonucu da proje tablolarına yaz (normalde engellenir)") + help="required: live makes real calls with the real key and spends quota") + ap.add_argument("--base", default=None, help="base URL (default depends on the mode)") + ap.add_argument("--phone", default=None, help="probe number (E.164)") + ap.add_argument("--sink", default=None, help="subscription webhook address (public HTTPS)") + ap.add_argument("--include-write", action="store_true", help="also try endpoints that create records (subscription, QoD)") + ap.add_argument("--only", default=None, help="comma-separated probe keys, e.g. location_verify,reachability") ap.add_argument("--timeout", type=float, default=8.0) ap.add_argument("--retries", type=int, default=1) args = ap.parse_args() - n = load_dotenv(ROOT / ".env") or load_dotenv(SHARED / ".env") - if n: - print(f".env → {n} değişken yüklendi") - - if args.mode == "live" and os.environ.get("NAC_OAUTH_TOKEN_URL") and not os.environ.get("NAC_OAUTH_TOKEN"): - try: - tok = fetch_oauth_token() - if tok: - os.environ["NAC_OAUTH_TOKEN"] = tok - print("OAuth2 client-credentials token alındı") - except Exception as e: - print(f"UYARI: token alınamadı ({type(e).__name__}: {e}) — passthrough uçları 401 dönecek") - - base = args.base or ( - "http://127.0.0.1:8081" if args.mode == "simulator" else "https://network-as-code.p-eu.rapidapi.com" - ) + env = load_env() + if env: + print(f".env loaded from {env}") + base = args.base or ("http://127.0.0.1:8081" if args.mode == "simulator" else "https://network-as-code.p-eu.rapidapi.com") cfg = NacConfig( mode=args.mode, base_url=base, rapidapi_key=os.environ.get("NAC_RAPIDAPI_KEY", ""), rapidapi_host=os.environ.get("NAC_RAPIDAPI_HOST", "network-as-code.nokia.rapidapi.com"), - oauth_token=os.environ.get("NAC_OAUTH_TOKEN", ""), timeout_s=args.timeout, retries=args.retries, - breaker_threshold=99, # probe: an error on one endpoint must not shut off the others + breaker_threshold=99, # a probe: one failing endpoint must not shut off the others ) if args.mode == "live" and not cfg.rapidapi_key: - print("ERROR: NAC_RAPIDAPI_KEY is not set. Create .env from .env.example, or rehearse with --mode simulator.", file=sys.stderr) + print("ERROR: NAC_RAPIDAPI_KEY is not set. Create .env from .env.example, or rehearse with --mode simulator.", + file=sys.stderr) return 2 - phone = args.phone or os.environ.get("NAC_PROBE_PHONE") or "+905551234567" - sink = args.sink or (os.environ.get("PUBLIC_BASE_URL", "").rstrip("/") + "/webhooks/probe" if os.environ.get("PUBLIC_BASE_URL") else "http://127.0.0.1:8000/webhooks/probe") + phone = args.phone or os.environ.get("NAC_PROBE_PHONE") or DEFAULT_PROBE_PHONE + public = os.environ.get("PUBLIC_BASE_URL", "").rstrip("/") + sink = args.sink or (f"{public}/webhooks/probe" if public else "http://127.0.0.1:8000/webhooks/probe") ctx = ProbeCtx(phone=phone, sink=sink) only = set(args.only.split(",")) if args.only else None started = datetime.now(timezone.utc).isoformat(timespec="seconds") - print(f"\nNaC sondası — mod={args.mode} taban={base} numara={mask_phone(phone)}") - print(f"{'-' * 78}") + print(f"\nNaC probe — mode={args.mode} base={base} number={mask_phone(phone)}") + print("-" * 78) client = NacClient(cfg) results = run_probes(client, ctx, args.include_write, only) cleaned = cleanup(client, ctx) if ctx.created else [] meta = {"started": started, "mode": args.mode, "base_url": base, "phone_masked": mask_phone(phone), - "sink": sink, "has_key": bool(cfg.rapidapi_key), "has_token": bool(cfg.oauth_token), - "include_write": args.include_write} + "sink": sink, "has_key": bool(cfg.rapidapi_key), "include_write": args.include_write} j, md = write_report(results, meta, cleaned) ok = sum(1 for r in results if r.get("ok")) reached = sum(1 for r in results if not r.get("ok") and not r.get("skipped") and endpoint_reached(r)) bad = sum(1 for r in results if not r.get("ok") and not r.get("skipped") and not endpoint_reached(r)) - sk = sum(1 for r in results if r.get("skipped")) - print(f"{'-' * 78}") - print(f"{ok} çalıştı · {reached} uç var (iş kuralı hatası) · {bad} erişilemedi · {sk} atlandı") - print(f"rapor: {md.relative_to(SHARED)}\n {j.relative_to(SHARED)}") - - if args.patch: - stamp = datetime.now().strftime("%d.%m.%Y") - # Safety lock: writing a non-live result into the "SIMULATOR planında test" column would be - # fabricated evidence. The master inventory table is written (its header states the mode); the - # project tables are not touched without --force. - table_dry = args.dry_run or (args.mode != "live" and not args.force) - if args.mode != "live" and not args.force: - print("\nKİLİT: mod canlı değil → proje tabloları DEĞİŞTİRİLMEDİ (uydurma kanıt yazmıyoruz).") - print(" Yine de ne yazılacağını görmek için: --force ekle.") - print("\napi-availability.md" + (" (yazılmıyor)" if table_dry else " güncelleniyor") + ":") - for name in PROJECTS: - r = patch_availability(ROOT / name, results, stamp, table_dry) - if r.get("error"): - print(f" {name:<14} ✖ {r['error']}") - else: - verb = "satır hazır (yazılmadı)" if table_dry else "satır yazıldı" - extra = f" · {len(r['skipped_rows'])} satır denenmedi" if r["skipped_rows"] else "" - extra += f" · TABLODA EŞLEŞMEYEN: {', '.join(r['unmatched'])}" if r["unmatched"] else "" - print(f" {name:<14} kolon={r['col']} · {len(r['patched'])} {verb}{extra}") - # The master inventory is written in every mode — its header states clearly which mode it ran in. - m = write_master_table(results, meta, stamp, args.dry_run) - print(f" ana envanter {m.relative_to(SHARED)}" + (" (dry-run)" if args.dry_run else "")) + skipped = sum(1 for r in results if r.get("skipped")) + print("-" * 78) + print(f"{ok} succeeded · {reached} endpoints exist (business-rule error) · {bad} unreachable · {skipped} skipped") + print(f"report: {md.relative_to(ROOT)}\n {j.relative_to(ROOT)}") return 0 if bad == 0 else 1 if __name__ == "__main__": - raise SystemExit(main()) + sys.exit(main()) From 0556c1c565a542b3a92dfeacc4450ec95be8332c Mon Sep 17 00:00:00 2001 From: Huseyin Kutsi Balci <262747299+kutsibalci@users.noreply.github.com> Date: Sun, 27 Sep 2026 15:13:09 +0300 Subject: [PATCH 2/2] Docs: the removed APIs, the tag for the 9 Sep probe, and "Top 10 finalist" - README and docs/api-availability.md explain that the 9 Sep 2026 probe ran with verify_live.py at tag v0.1-hackathon, which also probed APIs since removed, so those results in evidence/ are historical and reproducible from the tag. - README keeps "Top 10 finalist" and drops "Results pending". - docs/api-availability.md: the QoD path is v1, as in the code and the probe, not v0. - DEVELOPER-GUIDE: the qod_create example now matches the real signature. - .env.example: the OAuth block is removed, since nothing reads those variables now. Co-Authored-By: Claude Opus 5.5 --- .env.example | 11 ----------- DEVELOPER-GUIDE.md | 7 ++++--- README.md | 10 ++++++++-- docs/api-availability.md | 5 +++-- 4 files changed, 15 insertions(+), 18 deletions(-) diff --git a/.env.example b/.env.example index 4764266..576f9ca 100644 --- a/.env.example +++ b/.env.example @@ -7,17 +7,6 @@ NAC_RAPIDAPI_KEY= NAC_RAPIDAPI_HOST=network-as-code.nokia.rapidapi.com -# --- Bearer for passthrough/ endpoints --------------------------------------- -# There are two ways; fill in one: -# (a) write the ready-made access token you got from the portal directly -NAC_OAUTH_TOKEN= -# (b) or provide the client-credentials details — the probe obtains the token itself. -# Copy the TOKEN_URL value from the application settings in the Nokia portal. -NAC_OAUTH_TOKEN_URL= -NAC_OAUTH_CLIENT_ID= -NAC_OAUTH_CLIENT_SECRET= -NAC_OAUTH_SCOPE= - # --- Probe settings ----------------------------------------------------------- # Check the portal documentation for which test numbers are defined on the SIMULATOR plan; # write one of them here. If empty, +99999000001000 is tried. diff --git a/DEVELOPER-GUIDE.md b/DEVELOPER-GUIDE.md index e18a22c..9f09937 100644 --- a/DEVELOPER-GUIDE.md +++ b/DEVELOPER-GUIDE.md @@ -63,9 +63,10 @@ nac.reachability("+99999000001000") nac.congestion_query("+99999000001000") nac.geofence_subscribe("+99999000001000", lat=25.38, lng=51.49, radius_m=500, sink="https://example.com/webhooks/geofence", types=["org.camaraproject.geofencing-subscriptions.v0.area-entered"]) -nac.qod_create("+99999000001000", profile="QOS_E", duration_s=600, app_server_ipv4="203.0.113.10") +nac.qod_create("+99999000001000", app_server_ipv4="203.0.113.10", qos_profile="QOS_E", duration_s=600) nac.location_retrieve("+99999000001000", max_age_s=60) # after escalation only — the one call that returns a coordinate ``` -Deliberately unused: SIM/Device Swap, the KYC family, Number Verification. HeatShield measures exposure; it does not -establish identity, and calling those would violate data minimisation. +Deliberately absent: SIM/Device Swap, the KYC family, Number Verification. HeatShield measures exposure; it does not +establish identity, and calling those would violate data minimisation — so the client does not contain them. The +9 Sep 2026 probe that also tried them can be re-run exactly from tag `v0.1-hackathon`. diff --git a/README.md b/README.md index c14cc5c..57787a9 100644 --- a/README.md +++ b/README.md @@ -50,8 +50,9 @@ All six are called by the prototype through a single entry point (`packages/nac_ | **Location Retrieval** | Where to send the medic — **after escalation only**, once, coarsened to three decimals before it enters the ledger. The only API here that returns a coordinate, so the most tightly gated. | **Network Slicing** is a separate Nokia product on the roadmap. It is marked `# MOCK:` in `packages/agent/policy.py` -and is **not** presented as live. Deliberately unused: SIM/Device Swap, the KYC family, Number Verification — +and is **not** presented as live. Deliberately absent: SIM/Device Swap, the KYC family, Number Verification — HeatShield does not establish identity, it measures exposure, and calling those would violate data minimisation. +The operator client contains only the calls the product makes. ## Running it @@ -194,6 +195,11 @@ Location Verification radius scan in [`evidence/partial-scan-20260909-124602.json`](evidence/partial-scan-20260909-124602.json), and the endpoint-by-endpoint record in [`docs/api-availability.md`](docs/api-availability.md). +The 9 Sep 2026 live probe in `evidence/` was run with `tools/verify_live.py` at tag +[`v0.1-hackathon`](https://github.com/kutsibalci/heatshield/tree/v0.1-hackathon). That version also probed APIs +the product never calls (KYC, SIM swap, number verification); those have since been removed from the codebase, so +their results in the evidence file are historical. The probe can be re-run exactly from that tag. + **On WBGT — and this is the biggest gap in the prototype.** The product does not measure WBGT, it consumes it. Today it consumes it **from outside**: the value is pushed in by `POST /v1/sites/{site_id}/wbgt` and the ledger records it as `source: meteorology-feed`. **No weather provider is integrated.** There is no code in this @@ -309,7 +315,7 @@ modes need no credentials at all — the demos and the full test suite run with ## Status and what comes next - **13 Sep 2026** — prototype submitted. **21 Sep 2026** — Demo Day: presented live to the jury as one of - ten finalists. Results pending. + ten finalists. - **Next, in this order:** 1. Measure what Quality on Demand actually adds on the escalation path — the jury asked, and we have not measured it yet. Until then it stays what it is today: outside the verification budget and outside the diff --git a/docs/api-availability.md b/docs/api-availability.md index 0c4e4df..d5c27d4 100644 --- a/docs/api-availability.md +++ b/docs/api-availability.md @@ -10,11 +10,12 @@ from the follow-up scans in the same folder. Nothing here is asserted from docum | Location Verification | `location-verification/v1` | **200** · well-formed verdict and `lastLocationTime` · `maxAge` is mandatory (422 without it) · in the sandbox the verdict is `FALSE` for every area, including a 200 km circle around the device's own coordinate — put to Nokia | First rung of the cost ladder: a verdict, never a coordinate; loose `maxAge` = cached, strict `maxAge` = fresh fix and liveness probe | | Device Reachability Status | `device-reachability-status/v1` | **200** · `reachable: true`, `connectivity: ["SMS"]`, `lastStatusTime` | Silent-device detection inside a breached zone; the status can lag, so a strict-freshness fix overrules it | | Congestion Insights | `congestion-insights/v0` | **200** · four-interval forecast, latest `High` at confidence 56 | Cluster test: several devices dark together in a congested cell is a network event, not a medical one | -| Quality on Demand | `quality-on-demand/v0` | **200** · session opened with `QOS_E`, 600 s, application server `203.0.113.10`; read back from the session list; deleted · `qosStatus` came back `REQUESTED`, not `AVAILABLE` | Guaranteed bandwidth for the medic-to-physician assessment, after escalation | +| Quality on Demand | `quality-on-demand/v1` | **200** · session opened with `QOS_E`, 600 s, application server `203.0.113.10`; read back from the session list; deleted · `qosStatus` came back `REQUESTED`, not `AVAILABLE` | Guaranteed bandwidth for the medic-to-physician assessment, after escalation | | Location Retrieval | `location-retrieval/v0` | **200** · coordinate returned (the sandbox device sits in Budapest) · platform-reported uncertainty radius **1000 m** | Once, after escalation, coarsened to three decimals before it enters the ledger | **Not called, on purpose:** Network Slicing is a separate Nokia product and appears in the code only as `# MOCK`. SIM and -Device Swap, KYC and Number Verification establish identity; HeatShield measures exposure and does not need them. +Device Swap, KYC and Number Verification establish identity; HeatShield measures exposure and does not need them — +the client no longer contains them at all. The 9 Sep 2026 live probe in `evidence/` was run with `tools/verify_live.py` at tag `v0.1-hackathon`. That version also probed APIs the product never calls (KYC, SIM swap, number verification); those have since been removed from the codebase, so their results in the evidence file are historical. **Not integrated:** a weather provider. The WBGT value is an input to `/v1/sites/{id}/wbgt`; the demos feed it from the scenario. Wiring a meteorological source is roadmap.