#!/usr/bin/env python3 """Verify confidential api.read record access and write denial on loopback D3. Creates a case-local Connected App and Account; deletes both and revokes issued credentials. The runtime env is used only for local browser bootstrap. Evidence contains statuses and assertions, never passwords, secrets, or tokens. """ import argparse import http.cookiejar import json import secrets import shlex 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, runtime_env, expected_read_status): 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 verifier accepts only an HTTP loopback origin") base = base.rstrip("/") env = {} for line in Path(runtime_env).read_text().splitlines(): if "=" in line and not line.lstrip().startswith("#"): key, value = line.split("=", 1) words = shlex.split(value) env[key] = words[0] if words else "" jar = http.cookiejar.CookieJar() browser = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect(), urllib.request.HTTPCookieProcessor(jar)) machine = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect()) checks, failures, issued = [], [], [] app_id = account_id = admin = None def request(path, *, token=None, method="GET", data=None, form=None, controlled=False): headers = {} if controlled: headers.update({"Origin": base, "X-NexusCRM-Session-Control-Intent": "use", "Sec-Fetch-Site": "same-origin"}) 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 = (browser if controlled else machine).open( urllib.request.Request(base + path, data=body, headers=headers, method=method), timeout=40) except urllib.error.HTTPError as error: response = error # Secure browser-control cookies are transported only to the validated # HTTP loopback development origin; machine requests never use this jar. if controlled: for cookie in jar: cookie.secure = False raw = response.read() try: payload = json.loads(raw) if raw else {} except ValueError: payload = {} return response.status, payload def check(name, actual, expected, required=False): checks.append({"check": name, "actual": actual, "expected": expected}) if actual != expected: failures.append(name) if required: raise RuntimeError("Fixture prerequisite failed: " + name) def issue(name, form, controlled=False): status, body = request("/services/oauth2/token", method="POST", form=form, controlled=controlled) for key in ("access_token", "refresh_token"): if body.get(key): issued.append(body[key]) check(name, status, 200, required=True) return body try: login = issue("browser_login", {"grant_type": "password", "client_id": "nexuscrm-ui", "client_secret": "nexuscrm-ui-internal", "username": env.get("VERIFY_USERNAME") or "admin@example.invalid", "password": env["NEXUS_ADMIN_PASSWORD"]}, controlled=True) admin = login["access_token"] status, user = request("/services/oauth2/userinfo", token=admin, controlled=True) check("browser_identity", status, 200, required=True) client, secret = "api-read-probe-" + secrets.token_hex(8), secrets.token_urlsafe(32) status, body = request("/services/data/v63.0/sobjects/_System_ConnectedApp", method="POST", token=admin, controlled=True, data={"Name": "Case-local api.read verifier", "ConsumerKey": client, "ConsumerSecret": secret, "ClientType": "confidential", "TokenEndpointAuthMethod": "client_secret_post", "Scopes": "api.read", "AllowedGrantTypes": "client_credentials", "IsActive": True, "PermittedUsers": "AllUsers", "RunAsUserId": user["user_id"]}) check("create_connected_app", status, 201, required=True) app_id = body["id"] token_response = issue("client_credentials", {"grant_type": "client_credentials", "client_id": client, "client_secret": secret, "scope": "api.read"}) read_token = token_response["access_token"] check("issued_scope", token_response.get("scope"), "api.read") check("no_refresh_token", bool(token_response.get("refresh_token")), False) status, describe = request("/services/data/v63.0/sobjects/Account/describe", token=admin, controlled=True) check("fixture_describe", status, 200, required=True) record_type = next(item["recordTypeId"] for item in describe["recordTypeInfos"] if item.get("active") and item.get("available") and item.get("defaultRecordTypeMapping")) name = "ApiReadProbe-" + secrets.token_hex(8) fixture = {"Name": name, "Industry": "Others", "NumberOfEmployees": 1, "RecordTypeId": record_type, "Region__c": "APAC", "Website": "https://example.invalid"} status, body = request("/services/data/v63.0/sobjects/Account", method="POST", token=admin, controlled=True, data=fixture) check("create_account", status, 201, required=True) account_id = body["id"] record_path = "/services/data/v63.0/sobjects/Account/" + account_id paths = [("list", "/services/data/v63.0/sobjects/"), ("record", record_path), ("queryAll", "/services/data/v63.0/queryAll?" + urllib.parse.urlencode({"q": f"SELECT Id FROM Account WHERE Id = '{account_id}'"}))] for obj in ("Account", "Lead", "Opportunity", "Contact", "Task", "Order"): paths.extend([("query_" + obj, "/services/data/v63.0/query?" + urllib.parse.urlencode({"q": f"SELECT Id FROM {obj} LIMIT 1"})), ("describe_" + obj, "/services/data/v63.0/sobjects/" + obj + "/describe")]) for label, path in paths: status, _ = request(path, token=admin, controlled=True) check("browser_" + label, status, 200) status, payload = request(path, token=read_token) check("api_read_" + label, status, expected_read_status) if status == 200 and label == "record": check("read_fixture_name", payload.get("Name"), name) if status == 200 and label == "queryAll": check("queryAll_fixture", [item.get("Id") for item in payload.get("records", [])], [account_id]) 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, body = request(path, method=method, data=data, token=read_token) if method == "POST" and status == 201: cleanup_status, _ = request("/services/data/v63.0/sobjects/Account/" + body["id"], method="DELETE", token=admin, controlled=True) check("unexpected_write_cleanup", cleanup_status, 204) check("deny_" + method, status, 403) status, body = request(record_path, token=admin, controlled=True) check("fixture_preserved", status, 200) check("fixture_unchanged", body.get("Name"), name) except Exception as error: failures.append(type(error).__name__) finally: def cleanup(name, path, expected, **kwargs): try: status, _ = request(path, **kwargs) check(name, status, expected) except Exception: check(name, "request_failed", expected) for label, obj, record_id in (("account", "Account", account_id), ("app", "_System_ConnectedApp", app_id)): if record_id: cleanup("cleanup_" + label, "/services/data/v63.0/sobjects/" + obj + "/" + record_id, 204, method="DELETE", token=admin, controlled=True) for index, credential in enumerate(reversed(issued)): cleanup("revoke_" + str(index), "/services/oauth2/revoke", 200, method="POST", form={"token": credential}, controlled=True) if "read_token" in locals(): cleanup("revoked_machine_token", "/services/data/v63.0/query?" + urllib.parse.urlencode({"q": "SELECT Id FROM Account LIMIT 1"}), 401, token=read_token) return {"schema": "nexuscrm.local-api-read-check/v1", "baseUrl": base, "expectedReadStatus": expected_read_status, "checks": checks, "failures": failures, "status": "failed" if failures else "passed"} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--base-url", default="http://127.0.0.1:3000") parser.add_argument("--runtime-env", type=Path, required=True) 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.runtime_env, args.expect_read_status) args.output.write_text(json.dumps(result, indent=2) + "\n") print(json.dumps({"status": result["status"], "checks": len(result["checks"]), "failures": result["failures"]})) raise SystemExit(0 if result["status"] == "passed" else 1) if __name__ == "__main__": main()