#!/usr/bin/env python3 """Exercise public PKCE and its read/write boundary on a local D3 cluster. Requires an existing browser session for a local test user with Account CRUD. Creates one case-local Account as that user and deletes it during cleanup. Never prints or writes bearer tokens, refresh tokens, or authorization codes. """ from __future__ import annotations import argparse import base64 import hashlib import json import secrets import urllib.error import urllib.parse import urllib.request from pathlib import Path class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): return None def verify(base: str, browser_token: str, client_id: str, expected_read_status: int) -> dict: parsed = urllib.parse.urlsplit(base) if parsed.scheme != "http" or parsed.hostname not in {"127.0.0.1", "localhost", "::1"} or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}: raise ValueError("This fixture verifier accepts only an HTTP loopback origin") base = base.rstrip("/") opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect()) checks = [] issued = [] refresh_tokens = [] fixture_id = None def request(path, *, method="GET", token=None, data=None, form=None): headers = {} if token: headers["Authorization"] = "Bearer " + token body = None if data is not None: body = json.dumps(data).encode() headers["Content-Type"] = "application/json" elif form is not None: body = urllib.parse.urlencode(form).encode() headers["Content-Type"] = "application/x-www-form-urlencoded" try: response = opener.open(urllib.request.Request(base + path, data=body, headers=headers, method=method), timeout=45) except urllib.error.HTTPError as error: response = error raw = response.read() try: payload = json.loads(raw) if raw else None except ValueError: payload = None return response.status, response.headers, payload def check(name, actual, expected): checks.append({"check": name, "actual": actual, "expected": expected}) if actual != expected: raise AssertionError(f"{name}: expected {expected}, got {actual}") try: status, _, _ = request("/services/oauth2/userinfo", token=browser_token) check("browser_session", status, 200) redirect = "http://127.0.0.1:1717/callback" verifier = secrets.token_urlsafe(48) challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode().rstrip("=") state = secrets.token_urlsafe(24) status, headers, _ = request("/services/oauth2/authorize?" + urllib.parse.urlencode({ "response_type": "code", "client_id": client_id, "redirect_uri": redirect, "scope": "external.records.read", "state": state, "code_challenge": challenge, "code_challenge_method": "S256", })) check("authorize", status, 302) handle = urllib.parse.parse_qs(urllib.parse.urlsplit(headers["Location"]).query)["continue"][0] status, _, result = request("/services/oauth2/authorize/resume", method="POST", token=browser_token, data={"continue": handle}) check("resume", status, 200) callback = urllib.parse.parse_qs(urllib.parse.urlsplit(result["redirect_url"]).query) check("state", callback["state"][0] == state, True) status, _, result = request("/services/oauth2/token", method="POST", form={ "grant_type": "authorization_code", "client_id": client_id, "redirect_uri": redirect, "code": callback["code"][0], "code_verifier": verifier, }) check("token", status, 200) issued.append(result["access_token"]) public_token = result["access_token"] refresh_token = result["refresh_token"] refresh_tokens.append(refresh_token) check("scope", result.get("scope"), "external.records.read") name = "PublicScopeProbe-" + secrets.token_hex(8) status, _, describe = request("/services/data/v63.0/sobjects/Account/describe", token=browser_token) check("fixture_describe", status, 200) record_type = next(r["recordTypeId"] for r in describe["recordTypeInfos"] if r.get("active") and r.get("available") and r.get("defaultRecordTypeMapping")) fixture = {"Name": name, "Industry": "Others", "NumberOfEmployees": 1, "RecordTypeId": record_type, "Region__c": "APAC", "Website": "https://example.invalid"} status, _, result = request("/services/data/v63.0/sobjects/Account", method="POST", token=browser_token, data=fixture) check("fixture_create_as_browser", status, 201) fixture_id = result["id"] reads = [("list", "/services/data/v63.0/sobjects/")] for obj in ("Account", "Opportunity", "Contact", "Task", "Order"): reads.append(("query_" + obj, "/services/data/v63.0/query?" + urllib.parse.urlencode({"q": f"SELECT Id FROM {obj} LIMIT 1"}))) reads.append(("describe_" + obj, "/services/data/v63.0/sobjects/" + obj + "/describe")) record_path = "/services/data/v63.0/sobjects/Account/" + fixture_id reads.append(("record", record_path)) reads.append(("queryAll", "/services/data/v63.0/queryAll?" + urllib.parse.urlencode({"q": f"SELECT Id FROM Account WHERE Id = '{fixture_id}'"}))) for label, path in reads: status, _, _ = request(path, token=browser_token) check("api_" + label, status, 200) status, _, _ = request(path, token=public_token) check("public_" + label, status, expected_read_status) for method, path, data in ( ("POST", "/services/data/v63.0/sobjects/Account", {**fixture, "Name": name + "-denied"}), ("PATCH", record_path, {"Name": name + "-changed"}), ("DELETE", record_path, None), ): status, _, result = request(path, method=method, token=public_token, data=data) if method == "POST" and status == 201: request("/services/data/v63.0/sobjects/Account/" + result["id"], method="DELETE", token=browser_token) check("denied_" + method, status, 403) status, _, result = request(record_path, token=browser_token) check("fixture_still_exists", status, 200) check("fixture_unchanged", result.get("Name"), name) status, _, result = request("/services/oauth2/token", method="POST", form={ "grant_type": "refresh_token", "client_id": client_id, "refresh_token": refresh_token, }) check("refresh", status, 200) issued.append(result["access_token"]) if result.get("refresh_token"): refresh_tokens.append(result["refresh_token"]) check("refreshed_scope", result.get("scope"), "external.records.read") status, _, _ = request(reads[1][1], token=result["access_token"]) check("refreshed_query", status, expected_read_status) finally: cleanup_errors = [] cleanup_requests = [ ("revoke_verification_session", "/services/oauth2/revoke", {"method": "POST", "form": {"token": token}}, 200) for token in dict.fromkeys([*issued, *refresh_tokens]) ] if fixture_id: cleanup_requests.append(("fixture_cleanup", "/services/data/v63.0/sobjects/Account/" + fixture_id, {"method": "DELETE", "token": browser_token}, 204)) for label, path, kwargs, expected in cleanup_requests: try: status, _, _ = request(path, **kwargs) checks.append({"check": label, "actual": status, "expected": expected}) if status != expected: cleanup_errors.append(f"{label}: HTTP {status}") except Exception: cleanup_errors.append(f"{label}: request failed") # A successful revoke response alone is not proof of invalidation. for token in dict.fromkeys(refresh_tokens): try: status, _, result = request("/services/oauth2/token", method="POST", form={ "grant_type": "refresh_token", "client_id": client_id, "refresh_token": token, }) denied = status == 400 and isinstance(result, dict) and result.get("error") == "invalid_grant" checks.append({"check": "revoked_refresh_denied", "actual": denied, "expected": True}) if not denied: # Clean any credential unexpectedly minted by the negative probe. if isinstance(result, dict): for key in ("access_token", "refresh_token"): if result.get(key): request("/services/oauth2/revoke", method="POST", form={"token": result[key]}) cleanup_errors.append("revoked refresh token was not rejected as invalid_grant") except Exception: cleanup_errors.append("revoked refresh verification failed") if cleanup_errors: raise AssertionError("Cleanup incomplete: " + "; ".join(cleanup_errors)) return {"schema": "nexuscrm.local-public-connected-app-check/v1", "baseUrl": base, "expectedReadStatus": expected_read_status, "checks": checks, "status": "passed"} def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--base-url", default="http://127.0.0.1:3000") parser.add_argument("--browser-token-file", type=Path, required=True) parser.add_argument("--client-id", default="5M-Jie12V1h65_qvE499b09PGMHH1yCXF_YvGnJ3N88") parser.add_argument("--expect-read-status", type=int, choices=(200, 403), default=200) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() result = verify(args.base_url, args.browser_token_file.read_text().strip(), args.client_id, args.expect_read_status) args.output.write_text(json.dumps(result, indent=2) + "\n") print(f"Public Connected App verification passed ({len(result['checks'])} checks)") if __name__ == "__main__": main()