#!/usr/bin/env python3 """Infer by Flow7 Codex Provider Doctor. Version 0.1.1, protocol record dated 2026-08-11. This standalone diagnostic uses only the Python standard library. It has no telemetry and never calls infer.flow7.org unless that is the endpoint you explicitly provide. In safe mode it does not read an API key or send a prompt. Paid checks require both --run-paid-checks and an interactive confirmation. """ from __future__ import annotations import argparse import hmac import json import os import re import socket import ssl import sys import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass from typing import BinaryIO, Iterable, Iterator, Optional, Sequence, Tuple VERSION = "0.1.1" PROTOCOL_DATE = "2026-08-11" USER_AGENT = f"Infer-Codex-Provider-Doctor/{VERSION}" MAX_BODY_BYTES = 512 * 1024 MAX_SSE_LINE_BYTES = 64 * 1024 MAX_OUTPUT_TOKENS = 96 PAID_CONFIRMATION = "RUN PAID CHECKS" PAID_CHECK_NAMES = ("response", "stream", "tools", "structured") LOCAL_HOSTS = {"localhost", "127.0.0.1", "::1"} ENV_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]") KEY_LIKE = re.compile( r"(?i)(?:bearer\s+)?(?:sk-[A-Za-z0-9_-]{8,}|rly_[A-Za-z0-9_-]{8,}|[A-Za-z0-9_-]{32,})" ) SECRET_PATH_LABEL = re.compile( r"(?i)(?:api[_-]?key|access[_-]?token|auth[_-]?token|bearer|password|secret)" ) RESPONSE_MARKER = "PROVIDER_DOCTOR_OK" STREAM_MARKER = "PROVIDER_DOCTOR_STREAM_OK" TOOL_MARKER = "PROVIDER_DOCTOR_TOOL_OK" STRUCTURED_MARKER = "PROVIDER_DOCTOR_JSON_OK" class DoctorError(RuntimeError): """Expected, user-facing diagnostic failure.""" @dataclass(frozen=True) class Endpoints: origin: str base_url: str responses_url: str models_url: str host: str port: int scheme: str @dataclass(frozen=True) class Result: status: str name: str summary: str indeterminate: bool = False @dataclass(frozen=True) class SSEInspection: event_types: Tuple[str, ...] completed: bool failed: bool saw_created: bool saw_text_delta: bool terminal_status: str reported_model: str marker_matches: bool sequence_monotonic: bool class NoRedirectHandler(urllib.request.HTTPRedirectHandler): """Never forward Authorization to a redirect target.""" def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001 return None def direct_opener() -> urllib.request.OpenerDirector: """Build an opener with redirects and environment proxies disabled.""" return urllib.request.build_opener( urllib.request.ProxyHandler({}), NoRedirectHandler(), ) def redact_text(value: object, limit: int = 180) -> str: """Return a single-line label with likely credentials removed.""" text = CONTROL_CHARS.sub(" ", str(value or "")).strip() text = KEY_LIKE.sub("[redacted]", text) return text[:limit] or "[missing]" def validate_model(value: str) -> str: model = value.strip() if not model or len(model) > 200 or CONTROL_CHARS.search(model): raise DoctorError("Model must be 1-200 printable characters.") return model def decode_url_component(value: str, label: str) -> str: """Decode nested percent escapes so hidden controls and secrets are rejected.""" decoded = value for _ in range(8): try: candidate = urllib.parse.unquote(decoded, errors="strict") except UnicodeDecodeError as exc: raise DoctorError(f"Base URL {label} contains invalid percent-encoded UTF-8.") from exc if candidate == decoded: return decoded decoded = candidate raise DoctorError(f"Base URL {label} is excessively percent-encoded.") def validate_base_url(value: str, allow_http_localhost: bool = False) -> Endpoints: raw = value.strip() if not raw or any(character.isspace() for character in raw): raise DoctorError("Base URL is required and cannot contain whitespace.") if CONTROL_CHARS.search(raw): raise DoctorError("Base URL cannot contain control characters.") decoded_url = decode_url_component(raw, "value") if CONTROL_CHARS.search(decoded_url): raise DoctorError("Base URL cannot contain percent-encoded control characters.") try: parsed = urllib.parse.urlsplit(raw) port = parsed.port except ValueError as exc: raise DoctorError("Base URL has an invalid port.") from exc if parsed.scheme not in {"https", "http"}: raise DoctorError("Base URL must use https.") if not parsed.hostname: raise DoctorError("Base URL must include a host.") if parsed.username is not None or parsed.password is not None: raise DoctorError("Put credentials in an environment variable, never in the URL.") if parsed.query or parsed.fragment: raise DoctorError("Base URL cannot contain a query string or fragment.") host = parsed.hostname.lower().rstrip(".") if parsed.scheme == "http" and not (allow_http_localhost and host in LOCAL_HOSTS): raise DoctorError("Plain HTTP is refused. Use https, or explicitly allow HTTP for localhost.") decoded_path = decode_url_component(parsed.path, "path") if CONTROL_CHARS.search(decoded_path): raise DoctorError("Base URL path cannot contain control characters.") if KEY_LIKE.search(decoded_path) or SECRET_PATH_LABEL.search(decoded_path): raise DoctorError("Base URL path cannot contain credential-like or secret-like content.") segments = [part for part in decoded_path.split("/") if part] if any(part in {".", ".."} for part in segments): raise DoctorError("Base URL path cannot contain dot segments.") path = parsed.path.rstrip("/") if path.endswith("/responses"): responses_path = path base_path = path[: -len("/responses")] else: base_path = path responses_path = f"{base_path}/responses" if base_path else "/responses" models_path = f"{base_path}/models" if base_path else "/models" clean_netloc = parsed.netloc origin = urllib.parse.urlunsplit((parsed.scheme, clean_netloc, "", "", "")) base_url = urllib.parse.urlunsplit((parsed.scheme, clean_netloc, base_path or "/", "", "")) responses_url = urllib.parse.urlunsplit((parsed.scheme, clean_netloc, responses_path, "", "")) models_url = urllib.parse.urlunsplit((parsed.scheme, clean_netloc, models_path, "", "")) return Endpoints( origin=origin, base_url=base_url, responses_url=responses_url, models_url=models_url, host=host, port=port or (443 if parsed.scheme == "https" else 80), scheme=parsed.scheme, ) def read_limited(stream: BinaryIO, limit: int = MAX_BODY_BYTES) -> bytes: payload = stream.read(limit + 1) if len(payload) > limit: raise DoctorError(f"Response exceeded the {limit}-byte diagnostic limit.") return payload def parse_json(payload: bytes) -> dict: try: value = json.loads(payload.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise DoctorError("Endpoint returned a non-JSON response body.") from exc if not isinstance(value, dict): raise DoctorError("Endpoint returned JSON, but the top level was not an object.") return value def http_diagnosis(status: int) -> str: if 300 <= status < 400: return "Redirect refused; fix the base URL. Authorization was not forwarded." if status == 400 or status == 422: return "Request schema was rejected; compare the provider with the Responses API fields used by this doctor." if status == 401: return "Credential was missing or rejected. Check the selected environment variable and provider key." if status == 403: return "Credential reached the endpoint but lacks permission for this model or operation." if status == 404: return "Responses path or requested model was not found. Check whether the base URL ends at the API version root." if status == 409: return "Provider reported a state conflict. No retry was attempted." if status == 429: return "Provider reported a rate, quota, credit, or concurrency limit. No retry was attempted." if 500 <= status: return "Provider or upstream failed. No retry was attempted." return f"Unexpected HTTP {status}. No response body was printed." def network_diagnosis(exc: BaseException) -> str: reason = getattr(exc, "reason", exc) if isinstance(reason, socket.gaierror): return "DNS lookup failed. Check the host name." if isinstance(reason, ssl.SSLCertVerificationError): return "TLS certificate validation failed." if isinstance(reason, (socket.timeout, TimeoutError)): return "Connection timed out. No retry was attempted." if isinstance(reason, ConnectionRefusedError): return "Connection was refused by the target host." return "Network connection failed. Raw exception text was suppressed." def paid_network_failure(name: str, exc: BaseException) -> Result: return indeterminate_paid_failure(name, network_diagnosis(exc)) def indeterminate_paid_failure(name: str, summary: str) -> Result: return Result( "fail", name, f"{summary} The request may have reached the provider and may have been charged; " "remaining paid probes were not sent.", indeterminate=True, ) def check_dns(endpoints: Endpoints, timeout: float) -> Result: previous_timeout = socket.getdefaulttimeout() socket.setdefaulttimeout(timeout) try: records = socket.getaddrinfo(endpoints.host, endpoints.port, type=socket.SOCK_STREAM) except OSError as exc: return Result("fail", "DNS", network_diagnosis(exc)) finally: socket.setdefaulttimeout(previous_timeout) unique = {(family, address[0]) for family, _, _, _, address in records} return Result("pass", "DNS", f"Resolved {len(unique)} address record(s); addresses are not printed.") def check_tls(endpoints: Endpoints, timeout: float) -> Result: if endpoints.scheme != "https": return Result("warn", "TLS", "Skipped for explicitly allowed localhost HTTP.") try: context = ssl.create_default_context() with socket.create_connection((endpoints.host, endpoints.port), timeout=timeout) as raw: with context.wrap_socket(raw, server_hostname=endpoints.host) as secure: protocol = redact_text(secure.version(), limit=32) except OSError as exc: return Result("fail", "TLS", network_diagnosis(exc)) return Result("pass", "TLS", f"Certificate and hostname validated; negotiated {protocol}.") def open_request(request: urllib.request.Request, timeout: float): # noqa: ANN201 return direct_opener().open(request, timeout=timeout) def check_models(endpoints: Endpoints, model: str, timeout: float) -> Result: request = urllib.request.Request( endpoints.models_url, headers={"Accept": "application/json", "User-Agent": USER_AGENT}, method="GET", ) try: with open_request(request, timeout) as response: status = int(response.status) payload = read_limited(response) except urllib.error.HTTPError as exc: status = int(exc.code) exc.close() if status in {401, 403}: return Result( "warn", "Unauthenticated model list", f"Host and models path returned HTTP {status}; this establishes only that the host/path responded. " "Authentication, authorization, model availability, and API compatibility remain unresolved.", ) level = "fail" if status in {404} or 300 <= status < 400 else "warn" return Result(level, "Unauthenticated model list", http_diagnosis(status)) except urllib.error.URLError as exc: return Result("fail", "Unauthenticated model list", network_diagnosis(exc)) except DoctorError as exc: return Result("warn", "Unauthenticated model list", str(exc)) if status != 200: return Result("warn", "Unauthenticated model list", http_diagnosis(status)) try: document = parse_json(payload) except DoctorError as exc: return Result("warn", "Unauthenticated model list", str(exc)) items = document.get("data") if not isinstance(items, list): return Result("warn", "Unauthenticated model list", "HTTP 200 JSON did not expose a data array.") identifiers = { item.get("id") for item in items if isinstance(item, dict) and isinstance(item.get("id"), str) } if model in identifiers: return Result("pass", "Unauthenticated model list", "Requested model is listed; the list itself is not printed.") return Result( "warn", "Unauthenticated model list", f"Endpoint returned {len(identifiers)} model identifier(s), but not the requested identifier.", ) def request_json_payload(model: str) -> dict: return { "model": model, "input": f"Reply with exactly {RESPONSE_MARKER} and nothing else.", "max_output_tokens": MAX_OUTPUT_TOKENS, "store": False, } def stream_json_payload(model: str) -> dict: return { "model": model, "input": f"Reply with exactly {STREAM_MARKER} and nothing else.", "max_output_tokens": MAX_OUTPUT_TOKENS, "store": False, "stream": True, } def tools_json_payload(model: str) -> dict: return { "model": model, "input": f"Call provider_doctor_echo with value {TOOL_MARKER}. Do not answer in text.", "max_output_tokens": MAX_OUTPUT_TOKENS, "store": False, "tools": [ { "type": "function", "name": "provider_doctor_echo", "description": "Return the fixed provider-doctor marker.", "parameters": { "type": "object", "properties": {"value": {"type": "string", "enum": [TOOL_MARKER]}}, "required": ["value"], "additionalProperties": False, }, "strict": True, } ], "tool_choice": {"type": "function", "name": "provider_doctor_echo"}, "parallel_tool_calls": False, } def structured_json_payload(model: str) -> dict: return { "model": model, "input": f"Return the fixed check value {STRUCTURED_MARKER}.", "max_output_tokens": MAX_OUTPUT_TOKENS, "store": False, "text": { "format": { "type": "json_schema", "name": "provider_doctor_result", "strict": True, "schema": { "type": "object", "properties": {"check": {"type": "string", "enum": [STRUCTURED_MARKER]}}, "required": ["check"], "additionalProperties": False, }, } }, } def paid_request( endpoints: Endpoints, key: str, payload: dict, timeout: float, accept: str = "application/json", ): # noqa: ANN201 encoded = json.dumps(payload, separators=(",", ":")).encode("utf-8") request = urllib.request.Request( endpoints.responses_url, data=encoded, headers={ "Accept": accept, "Authorization": f"Bearer {key}", "Content-Type": "application/json", "User-Agent": USER_AGENT, }, method="POST", ) return open_request(request, timeout) def extract_output_text(document: dict) -> str: chunks = [] for item in document.get("output", []): if not isinstance(item, dict) or item.get("type") != "message": continue for part in item.get("content", []): if isinstance(part, dict) and part.get("type") == "output_text" and isinstance(part.get("text"), str): chunks.append(part["text"]) return "".join(chunks) def identity_result(name: str, requested: str, reported: object) -> Result: if not isinstance(reported, str) or not reported.strip(): return Result("fail", name, "Response omitted the model field required for a useful identity record.") clean_reported = redact_text(reported, limit=200) if reported == requested: return Result("pass", name, f"Provider self-reported the requested model: {clean_reported}.") return Result( "warn", name, f"Provider self-reported {clean_reported}, not the requested identifier. This can be alias resolution or route drift.", ) def run_response_check(endpoints: Endpoints, model: str, key: str, timeout: float) -> Sequence[Result]: try: with paid_request(endpoints, key, request_json_payload(model), timeout) as response: status = int(response.status) payload = read_limited(response) except urllib.error.HTTPError as exc: status = int(exc.code) exc.close() return [Result("fail", "Responses JSON", http_diagnosis(status))] except urllib.error.URLError as exc: return [paid_network_failure("Responses JSON", exc)] except OSError as exc: return [paid_network_failure("Responses JSON", exc)] except DoctorError as exc: return [Result("fail", "Responses JSON", str(exc))] if status != 200: return [Result("fail", "Responses JSON", http_diagnosis(status))] try: document = parse_json(payload) except DoctorError as exc: return [Result("fail", "Responses JSON", str(exc))] if document.get("object") != "response" or document.get("status") != "completed": behavior = Result("fail", "Responses JSON", "HTTP 200 did not contain a completed Response object.") else: output = extract_output_text(document).strip() behavior = Result( "pass" if hmac.compare_digest(output, RESPONSE_MARKER) else "warn", "Responses JSON", "Completed response matched the fixed probe; output was discarded." if hmac.compare_digest(output, RESPONSE_MARKER) else "Response completed, but fixed-output semantics did not match; output was discarded.", ) return [behavior, identity_result("Model identity (JSON)", model, document.get("model"))] def iter_sse_events(stream: BinaryIO) -> Iterator[Tuple[str, str]]: event_name = "" data_lines = [] total = 0 while True: raw = stream.readline(MAX_SSE_LINE_BYTES + 1) if not raw: if data_lines: yield event_name, "\n".join(data_lines) return total += len(raw) if total > MAX_BODY_BYTES: raise DoctorError(f"SSE stream exceeded the {MAX_BODY_BYTES}-byte diagnostic limit.") if len(raw) > MAX_SSE_LINE_BYTES: raise DoctorError("SSE event line exceeded the diagnostic limit.") try: line = raw.decode("utf-8").rstrip("\r\n") except UnicodeDecodeError as exc: raise DoctorError("SSE stream was not valid UTF-8.") from exc if line == "": if data_lines: yield event_name, "\n".join(data_lines) event_name = "" data_lines = [] continue if line.startswith(":"): continue field, separator, value = line.partition(":") if separator and value.startswith(" "): value = value[1:] if field == "event": event_name = value elif field == "data": data_lines.append(value) def inspect_sse(stream: BinaryIO) -> SSEInspection: event_types = [] completed = False failed = False saw_created = False saw_text_delta = False terminal_status = "" reported_model = "" text_parts = [] sequences = [] for event_name, data in iter_sse_events(stream): if data == "[DONE]": continue try: payload = json.loads(data) except json.JSONDecodeError as exc: raise DoctorError("SSE data field was not valid JSON.") from exc if not isinstance(payload, dict): raise DoctorError("SSE data field was not a JSON object.") event_type = payload.get("type") or event_name if not isinstance(event_type, str) or not event_type: raise DoctorError("SSE event omitted its type.") event_types.append(event_type) sequence = payload.get("sequence_number") if isinstance(sequence, int): sequences.append(sequence) if event_type == "response.created": saw_created = True if event_type == "response.output_text.delta" and isinstance(payload.get("delta"), str): saw_text_delta = True if sum(len(part) for part in text_parts) < 512: text_parts.append(payload["delta"]) if event_type == "response.completed": completed = True terminal = payload.get("response") if isinstance(terminal, dict): terminal_status = str(terminal.get("status") or "") if isinstance(terminal.get("model"), str): reported_model = terminal["model"] if not text_parts: text_parts.append(extract_output_text(terminal)) if event_type in {"response.failed", "response.incomplete"}: failed = True sequence_monotonic = bool(sequences) and all( current > previous for previous, current in zip(sequences, sequences[1:]) ) marker_matches = hmac.compare_digest("".join(text_parts).strip(), STREAM_MARKER) return SSEInspection( event_types=tuple(event_types), completed=completed, failed=failed, saw_created=saw_created, saw_text_delta=saw_text_delta, terminal_status=terminal_status, reported_model=reported_model, marker_matches=marker_matches, sequence_monotonic=sequence_monotonic, ) def run_stream_check(endpoints: Endpoints, model: str, key: str, timeout: float) -> Sequence[Result]: try: with paid_request(endpoints, key, stream_json_payload(model), timeout, "text/event-stream") as response: status = int(response.status) content_type = str(response.headers.get("Content-Type", "")).lower() if status != 200: return [Result("fail", "Responses SSE", http_diagnosis(status))] inspection = inspect_sse(response) except urllib.error.HTTPError as exc: status = int(exc.code) exc.close() return [Result("fail", "Responses SSE", http_diagnosis(status))] except urllib.error.URLError as exc: return [paid_network_failure("Responses SSE", exc)] except OSError as exc: return [paid_network_failure("Responses SSE", exc)] except DoctorError as exc: return [Result("fail", "Responses SSE", str(exc))] if "text/event-stream" not in content_type: stream_result = Result("fail", "Responses SSE", "HTTP 200 did not declare text/event-stream.") elif inspection.failed: stream_result = Result("fail", "Responses SSE", "Stream emitted a failed or incomplete terminal event.") elif not inspection.completed or inspection.terminal_status != "completed": stream_result = indeterminate_paid_failure( "Responses SSE", "Stream closed without a completed response.completed event; Codex may report a disconnected stream.", ) elif not inspection.sequence_monotonic: stream_result = Result("fail", "Responses SSE", "SSE sequence_number values were missing or not strictly increasing.") elif not inspection.saw_created: stream_result = Result("warn", "Responses SSE", "Stream completed but omitted response.created.") elif not inspection.saw_text_delta or not inspection.marker_matches: stream_result = Result( "warn", "Responses SSE", "Stream completed, but delta/fixed-output semantics did not match; output was discarded.", ) else: stream_result = Result("pass", "Responses SSE", "Observed response.created, text delta, and response.completed; output was discarded.") return [stream_result, identity_result("Model identity (SSE)", model, inspection.reported_model)] def run_tools_check(endpoints: Endpoints, model: str, key: str, timeout: float) -> Sequence[Result]: try: with paid_request(endpoints, key, tools_json_payload(model), timeout) as response: status = int(response.status) payload = read_limited(response) except urllib.error.HTTPError as exc: status = int(exc.code) exc.close() return [Result("fail", "Function tool semantics", http_diagnosis(status))] except urllib.error.URLError as exc: return [paid_network_failure("Function tool semantics", exc)] except OSError as exc: return [paid_network_failure("Function tool semantics", exc)] except DoctorError as exc: return [Result("fail", "Function tool semantics", str(exc))] if status != 200: return [Result("fail", "Function tool semantics", http_diagnosis(status))] try: document = parse_json(payload) except DoctorError as exc: return [Result("fail", "Function tool semantics", str(exc))] calls = [item for item in document.get("output", []) if isinstance(item, dict) and item.get("type") == "function_call"] valid = False for call in calls: if call.get("name") != "provider_doctor_echo" or not isinstance(call.get("arguments"), str): continue try: arguments = json.loads(call["arguments"]) except json.JSONDecodeError: continue if arguments == {"value": TOOL_MARKER} and isinstance(call.get("call_id"), str) and call["call_id"]: valid = True break result = Result( "pass" if valid else "fail", "Function tool semantics", "Forced strict function call returned the exact arguments and a call_id; no tool was executed." if valid else "Response did not contain the forced strict function call, exact arguments, and call_id.", ) return [result, identity_result("Model identity (tools)", model, document.get("model"))] def run_structured_check(endpoints: Endpoints, model: str, key: str, timeout: float) -> Sequence[Result]: try: with paid_request(endpoints, key, structured_json_payload(model), timeout) as response: status = int(response.status) payload = read_limited(response) except urllib.error.HTTPError as exc: status = int(exc.code) exc.close() return [Result("fail", "Structured output semantics", http_diagnosis(status))] except urllib.error.URLError as exc: return [paid_network_failure("Structured output semantics", exc)] except OSError as exc: return [paid_network_failure("Structured output semantics", exc)] except DoctorError as exc: return [Result("fail", "Structured output semantics", str(exc))] if status != 200: return [Result("fail", "Structured output semantics", http_diagnosis(status))] document = {} try: document = parse_json(payload) structured = json.loads(extract_output_text(document)) except (DoctorError, json.JSONDecodeError): valid = False else: valid = structured == {"check": STRUCTURED_MARKER} result = Result( "pass" if valid else "fail", "Structured output semantics", "Strict JSON Schema output matched exactly; output was discarded." if valid else "Output did not match the exact strict JSON Schema; output was discarded.", ) return [result, identity_result("Model identity (structured)", model, document.get("model"))] def parse_paid_checks(value: str) -> Tuple[str, ...]: requested = tuple(part.strip().lower() for part in value.split(",") if part.strip()) unknown = [item for item in requested if item not in PAID_CHECK_NAMES] if not requested or unknown: allowed = ", ".join(PAID_CHECK_NAMES) raise argparse.ArgumentTypeError(f"Choose a comma-separated subset of: {allowed}.") return tuple(dict.fromkeys(requested)) def confirm_paid_checks(endpoint: Endpoints, model: str, checks: Sequence[str]) -> bool: print("\nPaid request boundary") print(f" Direct target: {endpoint.responses_url}") print(f" Requested model: {redact_text(model, 200)}") print(f" Requests: {len(checks)} ({', '.join(checks)})") print(f" Output cap: {MAX_OUTPUT_TOKENS} tokens per request") print(" store=false, no redirects, no proxy, no retries, no response text printed") print(" Your provider controls pricing and may still retain request data.") if not sys.stdin.isatty(): raise DoctorError("Paid checks require an interactive terminal confirmation.") answer = input(f'Type "{PAID_CONFIRMATION}" to continue: ').strip() return hmac.compare_digest(answer, PAID_CONFIRMATION) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Diagnose an OpenAI Responses-compatible Codex custom provider without relaying data through Infer.", ) parser.add_argument("--version", action="version", version=f"%(prog)s {VERSION} ({PROTOCOL_DATE})") parser.add_argument("--base-url", required=True, help="API version root, for example https://provider.example/v1") parser.add_argument("--model", required=True, help="Exact model identifier Codex will request") parser.add_argument("--timeout", type=float, default=20.0, help="Per-operation timeout in seconds (default: 20)") parser.add_argument("--offline", action="store_true", help="Validate inputs only; make no network connection") parser.add_argument( "--allow-http-localhost", action="store_true", help="Allow plain HTTP only for localhost/127.0.0.1/::1", ) parser.add_argument( "--run-paid-checks", action="store_true", help="Offer bounded Responses, SSE, function-tool, and structured-output probes after typed confirmation", ) parser.add_argument( "--paid-checks", type=parse_paid_checks, default=PAID_CHECK_NAMES, help="Comma-separated paid checks: response,stream,tools,structured", ) parser.add_argument( "--api-key-env", default="PROVIDER_API_KEY", help="Environment variable containing the provider key (default: PROVIDER_API_KEY)", ) return parser def print_result(result: Result) -> None: label = {"pass": "PASS", "warn": "WARN", "fail": "FAIL"}.get(result.status, "INFO") print(f"[{label}] {result.name}: {result.summary}") def run_paid(endpoints: Endpoints, model: str, key: str, timeout: float, checks: Iterable[str]) -> Sequence[Result]: runners = { "response": run_response_check, "stream": run_stream_check, "tools": run_tools_check, "structured": run_structured_check, } results = [] for name in checks: check_results = runners[name](endpoints, model, key, timeout) results.extend(check_results) if any(result.indeterminate for result in check_results): break return results def main(argv: Optional[Sequence[str]] = None) -> int: parser = build_parser() args = parser.parse_args(argv) try: if not 1 <= args.timeout <= 120: raise DoctorError("Timeout must be between 1 and 120 seconds.") if not ENV_NAME.fullmatch(args.api_key_env): raise DoctorError("API key environment-variable name is invalid.") model = validate_model(args.model) endpoints = validate_base_url(args.base_url, args.allow_http_localhost) if args.offline and args.run_paid_checks: raise DoctorError("--offline cannot be combined with --run-paid-checks.") except DoctorError as exc: parser.error(str(exc)) print(f"Infer by Flow7 Codex Provider Doctor {VERSION} — protocol record {PROTOCOL_DATE}") print("No telemetry. Raw model output and error bodies are never printed or written.") print(f"Target origin: {endpoints.origin}") print(f"Responses path: {urllib.parse.urlsplit(endpoints.responses_url).path}") print(f"Requested model: {redact_text(model, 200)}") results = [ Result( "pass", "URL policy", "URL contains no userinfo, query, fragment, secret-like path content, control bytes, or unsafe remote HTTP.", ) ] if args.offline: results.append(Result("pass", "Offline boundary", "No network connection was made and no credential was read.")) else: results.append(check_dns(endpoints, args.timeout)) results.append(check_tls(endpoints, args.timeout)) results.append(check_models(endpoints, model, args.timeout)) for result in results: print_result(result) if args.run_paid_checks: try: confirmed = confirm_paid_checks(endpoints, model, args.paid_checks) except DoctorError as exc: print(f"[BLOCKED] Paid checks: {exc}", file=sys.stderr) return 2 if not confirmed: print("[BLOCKED] Paid checks: confirmation did not match; no credential was read and no prompt was sent.") return 2 key = os.environ.get(args.api_key_env, "") if not key or CONTROL_CHARS.search(key): print( f"[BLOCKED] Paid checks: {args.api_key_env} is missing or invalid. The value was not printed.", file=sys.stderr, ) return 2 paid_results = run_paid(endpoints, model, key, args.timeout, args.paid_checks) results.extend(paid_results) for result in paid_results: print_result(result) else: print("[SAFE] Paid checks were not requested; the API-key environment variable was not read.") counts = {status: sum(result.status == status for result in results) for status in ("pass", "warn", "fail")} print(f"\nSummary: {counts['pass']} pass, {counts['warn']} warning, {counts['fail']} fail") print("A passing probe is not full Codex certification or proof of model weights, pricing, retention, or route origin.") return 1 if counts["fail"] else 0 if __name__ == "__main__": raise SystemExit(main())