| 1 | """Permission preflight contract and human renderer.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from typing import Any |
| 6 | |
| 7 | from . import env |
| 8 | |
| 9 | |
| 10 | ENDPOINT_OVERRIDE_KEYS = { |
| 11 | "BSKY_SEARCH_HOST", |
| 12 | "LAST30DAYS_SEARXNG_URL", |
| 13 | "LAST30DAYS_YOUTUBE_SSH_HOST", |
| 14 | "OPENAI_BASE_URL", |
| 15 | "XAI_BASE_URL", |
| 16 | "XIAOHONGSHU_API_BASE", |
| 17 | } |
| 18 | |
| 19 | PROVIDER_CREDENTIALS = { |
| 20 | "google": "Google/Gemini API key", |
| 21 | "openai": "OpenAI API key", |
| 22 | "xai": "xAI API key", |
| 23 | "openrouter": "OpenRouter API key", |
| 24 | "perplexity": "Perplexity API key", |
| 25 | "scrapecreators": "ScrapeCreators API key", |
| 26 | "github": "GitHub token or gh auth", |
| 27 | # X API v2 app-only bearer (X_BEARER_TOKEN). Presence is computed from |
| 28 | # config inside build(), never through diagnose.providers, whose key set |
| 29 | # is frozen by tests/test_diagnose_compat.py. |
| 30 | "x_bearer": "X API bearer token", |
| 31 | } |
| 32 | |
| 33 | |
| 34 | def _truthy(value: Any) -> bool: |
| 35 | if value is None: |
| 36 | return False |
| 37 | return str(value).strip().lower() in {"1", "true", "yes", "on"} |
| 38 | |
| 39 | |
| 40 | def _status(value: bool) -> str: |
| 41 | return "available" if value else "unavailable" |
| 42 | |
| 43 | |
| 44 | def _write_key(write: dict[str, str]) -> tuple[str, str]: |
| 45 | return str(write.get("kind") or ""), str(write.get("path") or "") |
| 46 | |
| 47 | |
| 48 | def _dedupe_writes(writes: list[dict[str, str]]) -> list[dict[str, str]]: |
| 49 | deduped: list[dict[str, str]] = [] |
| 50 | seen: set[tuple[str, str]] = set() |
| 51 | for write in writes: |
| 52 | key = _write_key(write) |
| 53 | if key in seen: |
| 54 | continue |
| 55 | seen.add(key) |
| 56 | deduped.append(write) |
| 57 | return deduped |
| 58 | |
| 59 | |
| 60 | def build( |
| 61 | config: dict[str, Any], |
| 62 | diagnose: dict[str, Any], |
| 63 | *, |
| 64 | planned_save_dir: str | None = None, |
| 65 | report_on_save_dir: str | None = None, |
| 66 | ) -> dict[str, Any]: |
| 67 | """Build a stable, secret-free permission preflight object.""" |
| 68 | browser = dict(diagnose.get("browser_cookies") or {}) |
| 69 | browser_mode = str(browser.get("mode") or "off") |
| 70 | browser_browsers = list(browser.get("browsers") or []) |
| 71 | browser_enabled = browser_mode in {"read", "plan_only"} and bool(browser_browsers) |
| 72 | if browser_enabled: |
| 73 | browser_status = "enabled_by_config" |
| 74 | else: |
| 75 | browser_status = "off" |
| 76 | |
| 77 | ignored_project_config = diagnose.get("ignored_project_config") |
| 78 | config_source = str(diagnose.get("config_source") or "env_only") |
| 79 | project_config_active = config_source.startswith("project:") |
| 80 | if project_config_active: |
| 81 | project_status = "trusted_active" |
| 82 | elif ignored_project_config: |
| 83 | project_status = "ignored_untrusted" |
| 84 | else: |
| 85 | project_status = "not_active" |
| 86 | |
| 87 | local_writes = list(diagnose.get("local_writes") or []) |
| 88 | if planned_save_dir: |
| 89 | local_writes = [{"kind": "report", "path": str(planned_save_dir)}] |
| 90 | local_writes = _dedupe_writes([dict(write) for write in local_writes]) |
| 91 | local_write_paths = {str(write.get("path") or "") for write in local_writes} |
| 92 | conditional_writes: list[dict[str, str]] = [] |
| 93 | if report_on_save_dir and not planned_save_dir and str(report_on_save_dir) not in local_write_paths: |
| 94 | conditional_writes.append({"kind": "report_on_save", "path": str(report_on_save_dir)}) |
| 95 | conditional_writes = _dedupe_writes(conditional_writes) |
| 96 | |
| 97 | providers = dict(diagnose.get("providers") or {}) |
| 98 | credentials = { |
| 99 | "google": {"present": bool(providers.get("google")), "label": PROVIDER_CREDENTIALS["google"]}, |
| 100 | "openai": {"present": bool(providers.get("openai")), "label": PROVIDER_CREDENTIALS["openai"]}, |
| 101 | "xai": {"present": bool(providers.get("xai")), "label": PROVIDER_CREDENTIALS["xai"]}, |
| 102 | "openrouter": {"present": bool(providers.get("openrouter")), "label": PROVIDER_CREDENTIALS["openrouter"]}, |
| 103 | "perplexity": {"present": bool(providers.get("perplexity")), "label": PROVIDER_CREDENTIALS["perplexity"]}, |
| 104 | "scrapecreators": { |
| 105 | "present": bool(diagnose.get("has_scrapecreators")), |
| 106 | "label": PROVIDER_CREDENTIALS["scrapecreators"], |
| 107 | }, |
| 108 | "github": {"present": bool(diagnose.get("has_github")), "label": PROVIDER_CREDENTIALS["github"]}, |
| 109 | "x_bearer": { |
| 110 | "present": bool(str(config.get("X_BEARER_TOKEN") or "").strip()), |
| 111 | "label": PROVIDER_CREDENTIALS["x_bearer"], |
| 112 | }, |
| 113 | } |
| 114 | |
| 115 | active_endpoint_overrides = sorted( |
| 116 | key for key in ENDPOINT_OVERRIDE_KEYS if config.get(key) |
| 117 | ) |
| 118 | ignored_endpoint_overrides = sorted(diagnose.get("ignored_endpoint_overrides") or []) |
| 119 | external_commands = { |
| 120 | name: {"status": _status(bool(available))} |
| 121 | for name, available in sorted((diagnose.get("external_commands") or {}).items()) |
| 122 | } |
| 123 | |
| 124 | action_items: list[str] = [] |
| 125 | if ignored_project_config: |
| 126 | action_items.append("Project config was ignored; set LAST30DAYS_TRUST_PROJECT_CONFIG=1 to trust it.") |
| 127 | # get_config() already emptied these, so the provider flags above read them |
| 128 | # as absent. Name them anyway: the user's setup is broken in a way the |
| 129 | # presence booleans alone describe as "nothing configured". |
| 130 | templated_keys = env.templated_config_keys(config) |
| 131 | if templated_keys: |
| 132 | action_items.append( |
| 133 | "Unsubstituted config template(s) count as unset: " |
| 134 | + _format_names(templated_keys) |
| 135 | + ". Replace each with a real value or remove it." |
| 136 | ) |
| 137 | |
| 138 | return { |
| 139 | "status": "action_needed" if action_items else "ready", |
| 140 | "safe": bool(diagnose.get("safe")), |
| 141 | "local_reads": { |
| 142 | "config_source": config_source, |
| 143 | "project_config": { |
| 144 | "status": project_status, |
| 145 | "trusted": bool(project_config_active), |
| 146 | "ignored_path": ignored_project_config, |
| 147 | "ignored_keys": list(diagnose.get("ignored_project_config_keys") or []), |
| 148 | }, |
| 149 | "browser_cookies": { |
| 150 | "status": browser_status, |
| 151 | "mode": browser_mode, |
| 152 | "browsers": browser_browsers, |
| 153 | "reads_values": False, |
| 154 | }, |
| 155 | }, |
| 156 | "local_writes": local_writes, |
| 157 | "conditional_writes": conditional_writes, |
| 158 | "external_commands": external_commands, |
| 159 | "credentials": credentials, |
| 160 | "network": { |
| 161 | "available_sources": list(diagnose.get("available_sources") or []), |
| 162 | "native_search": bool(diagnose.get("native_search")), |
| 163 | "endpoint_overrides": active_endpoint_overrides, |
| 164 | "ignored_endpoint_overrides": ignored_endpoint_overrides, |
| 165 | }, |
| 166 | "action_items": action_items, |
| 167 | } |
| 168 | |
| 169 | |
| 170 | def _format_names(names: list[str]) -> str: |
| 171 | return ", ".join(names) if names else "none" |
| 172 | |
| 173 | |
| 174 | def render_text(preflight: dict[str, Any]) -> str: |
| 175 | """Render the permission preflight as concise user-facing text.""" |
| 176 | lines: list[str] = ["last30days preflight"] |
| 177 | status = preflight.get("status") |
| 178 | if status == "ready": |
| 179 | lines.append("Status: Ready to research with safe defaults.") |
| 180 | else: |
| 181 | lines.append("Status: Ready, with item(s) to review.") |
| 182 | |
| 183 | reads = preflight.get("local_reads") or {} |
| 184 | project = reads.get("project_config") or {} |
| 185 | browser = reads.get("browser_cookies") or {} |
| 186 | writes = list(preflight.get("local_writes") or []) |
| 187 | conditional_writes = list(preflight.get("conditional_writes") or []) |
| 188 | commands = preflight.get("external_commands") or {} |
| 189 | credentials = preflight.get("credentials") or {} |
| 190 | network = preflight.get("network") or {} |
| 191 | |
| 192 | lines.append("") |
| 193 | lines.append("Local reads:") |
| 194 | lines.append(f"- Config source: {reads.get('config_source') or 'env_only'}") |
| 195 | if project.get("status") == "ignored_untrusted": |
| 196 | ignored_keys = _format_names(list(project.get("ignored_keys") or [])) |
| 197 | lines.append(f"- Project config: ignored untrusted file ({ignored_keys})") |
| 198 | elif project.get("status") == "trusted_active": |
| 199 | lines.append("- Project config: trusted and active") |
| 200 | else: |
| 201 | lines.append("- Project config: not active") |
| 202 | if browser.get("status") == "enabled_by_config": |
| 203 | lines.append( |
| 204 | "- Browser cookies: enabled by config for " |
| 205 | + _format_names(list(browser.get("browsers") or [])) |
| 206 | + "; preflight did not read cookie values" |
| 207 | ) |
| 208 | else: |
| 209 | lines.append("- Browser cookies: off; no browser stores will be read") |
| 210 | |
| 211 | lines.append("") |
| 212 | lines.append("Local writes:") |
| 213 | if writes: |
| 214 | for write in writes: |
| 215 | lines.append(f"- {write.get('kind', 'file')}: {write.get('path')}") |
| 216 | else: |
| 217 | lines.append("- none planned") |
| 218 | for write in conditional_writes: |
| 219 | if write.get("kind") == "report_on_save": |
| 220 | lines.append(f"- Report (if saved): {write.get('path')}") |
| 221 | else: |
| 222 | lines.append(f"- {write.get('kind', 'file')} (conditional): {write.get('path')}") |
| 223 | |
| 224 | present_credentials = [ |
| 225 | str(info.get("label") or name) |
| 226 | for name, info in credentials.items() |
| 227 | if info.get("present") |
| 228 | ] |
| 229 | lines.append("") |
| 230 | lines.append("Credentials:") |
| 231 | lines.append("- Present: " + _format_names(present_credentials)) |
| 232 | lines.append("- Values are not printed or written by preflight") |
| 233 | |
| 234 | unavailable_commands = [ |
| 235 | name for name, info in commands.items() if info.get("status") == "unavailable" |
| 236 | ] |
| 237 | lines.append("") |
| 238 | if unavailable_commands: |
| 239 | lines.append("Optional commands unavailable: " + _format_names(unavailable_commands)) |
| 240 | else: |
| 241 | lines.append("Optional commands: available") |
| 242 | |
| 243 | endpoint_overrides = list(network.get("endpoint_overrides") or []) |
| 244 | ignored_endpoint_overrides = list(network.get("ignored_endpoint_overrides") or []) |
| 245 | lines.append("") |
| 246 | lines.append("Network:") |
| 247 | lines.append("- Available sources: " + _format_names(list(network.get("available_sources") or []))) |
| 248 | if endpoint_overrides: |
| 249 | lines.append("- Endpoint overrides active: " + _format_names(endpoint_overrides)) |
| 250 | if ignored_endpoint_overrides: |
| 251 | lines.append("- Endpoint overrides ignored: " + _format_names(ignored_endpoint_overrides)) |
| 252 | |
| 253 | action_items = list(preflight.get("action_items") or []) |
| 254 | lines.append("") |
| 255 | if action_items: |
| 256 | lines.append("Next:") |
| 257 | for item in action_items: |
| 258 | lines.append(f"- {item}") |
| 259 | else: |
| 260 | lines.append("Next: run research normally, or configure optional sources if you need more coverage.") |
| 261 | |
| 262 | return "\n".join(lines) + "\n" |
| 263 |