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