1210 lines
56 KiB
Python
1210 lines
56 KiB
Python
"""GitHub Repo Inspector A2A agent.
|
|
|
|
The agent uses direct GitHub REST calls plus deterministic repository heuristics
|
|
so findings are evidence-backed and cite exact files/lines. A platform-scoped
|
|
LLM grant is still declared/read through ctx.llm per hosted-agent convention,
|
|
but no provider keys or LiteLLM environment variables are read directly.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import mimetypes
|
|
import re
|
|
from collections import Counter, defaultdict
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from pydantic import BaseModel, Field
|
|
|
|
from a2a_pack import (
|
|
A2AAgent,
|
|
EgressPolicy,
|
|
LLMProvisioning,
|
|
NoAuth,
|
|
Pricing,
|
|
Resources,
|
|
RunContext,
|
|
WorkspaceAccess,
|
|
WorkspaceMode,
|
|
skill,
|
|
)
|
|
from a2a_pack.context import LLMCreds
|
|
|
|
|
|
GITHUB_API = "https://api.github.com"
|
|
RUNTIME_SKILLS_DIR = "github-repo-inspector/.deepagents/skills/"
|
|
DEEPAGENTS_RECURSION_LIMIT = 500
|
|
|
|
CATEGORY_ORDER = [
|
|
"source",
|
|
"config",
|
|
"CI/CD",
|
|
"Docker/container",
|
|
"tests",
|
|
"docs",
|
|
"package/dependency",
|
|
"infra/IaC",
|
|
"scripts",
|
|
"hidden/dotfiles",
|
|
"assets/other",
|
|
]
|
|
|
|
VENDOR_BUILD_CACHE_DIRS = {
|
|
".git",
|
|
".hg",
|
|
".svn",
|
|
".next",
|
|
".nuxt",
|
|
".parcel-cache",
|
|
".pytest_cache",
|
|
".ruff_cache",
|
|
".tox",
|
|
".venv",
|
|
"venv",
|
|
"env",
|
|
"node_modules",
|
|
"bower_components",
|
|
"vendor",
|
|
"dist",
|
|
"build",
|
|
"target",
|
|
"coverage",
|
|
".coverage",
|
|
"__pycache__",
|
|
".mypy_cache",
|
|
".gradle",
|
|
".idea",
|
|
".vscode",
|
|
"out",
|
|
"tmp",
|
|
"temp",
|
|
"cache",
|
|
".cache",
|
|
}
|
|
|
|
BINARY_EXTENSIONS = {
|
|
".png", ".jpg", ".jpeg", ".gif", ".webp", ".ico", ".pdf", ".zip", ".gz",
|
|
".tgz", ".bz2", ".xz", ".7z", ".rar", ".jar", ".war", ".ear", ".class",
|
|
".pyc", ".pyo", ".so", ".dylib", ".dll", ".exe", ".bin", ".wasm", ".woff",
|
|
".woff2", ".ttf", ".otf", ".mp3", ".mp4", ".mov", ".avi", ".mkv", ".wav",
|
|
".flac", ".sqlite", ".db", ".parquet", ".avif", ".heic",
|
|
}
|
|
|
|
TEXT_EXTENSIONS = {
|
|
".py", ".js", ".jsx", ".ts", ".tsx", ".mjs", ".cjs", ".java", ".go", ".rs",
|
|
".rb", ".php", ".cs", ".cpp", ".cc", ".c", ".h", ".hpp", ".swift", ".kt",
|
|
".kts", ".scala", ".sh", ".bash", ".zsh", ".fish", ".ps1", ".bat", ".cmd",
|
|
".yml", ".yaml", ".json", ".toml", ".ini", ".cfg", ".conf", ".xml", ".html",
|
|
".css", ".scss", ".md", ".rst", ".txt", ".dockerfile", ".tf", ".tfvars", ".hcl",
|
|
".sql", ".graphql", ".proto", ".gradle", ".properties", ".env", ".example",
|
|
".lock", ".mod", ".sum", ".mk", ".cmake",
|
|
}
|
|
|
|
SECRET_REGEXES: list[tuple[str, re.Pattern[str]]] = [
|
|
("GitHub token", re.compile(r"gh[pousr]_[A-Za-z0-9_]{20,}")),
|
|
("AWS access key", re.compile(r"AKIA[0-9A-Z]{16}")),
|
|
("Private key", re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH |DSA )?PRIVATE KEY-----")),
|
|
("Slack token", re.compile(r"xox[baprs]-[A-Za-z0-9-]{20,}")),
|
|
("Generic assigned secret", re.compile(r"(?i)(api[_-]?key|secret|token|password|passwd|pwd)\s*[:=]\s*['\"]?([^'\"\s]{12,})")),
|
|
]
|
|
|
|
TODO_RE = re.compile(r"\b(TODO|FIXME|HACK|XXX)\b", re.IGNORECASE)
|
|
COMMENTED_BLOCK_RE = re.compile(r"^\s*(//|#|--|/\*|\*)\s*(if |for |while |def |function |class |const |let |var |return |import |from )")
|
|
RISKY_PACKAGE_HINTS = {"request", "left-pad", "event-stream", "ua-parser-js", "colors", "faker"}
|
|
|
|
|
|
class GithubRepoInspectorConfig(BaseModel):
|
|
github_api_base: str = GITHUB_API
|
|
|
|
|
|
class GithubRepoInspector(A2AAgent[GithubRepoInspectorConfig, NoAuth]):
|
|
name = "github-repo-inspector"
|
|
description = (
|
|
"GitHub Repo Inspector lists accessible repositories, inspects complete "
|
|
"GitHub repository trees with file-level citations, and can create "
|
|
"GitHub issues for selected recommendations."
|
|
)
|
|
version = "1.0.0"
|
|
|
|
config_model = GithubRepoInspectorConfig
|
|
auth_model = NoAuth
|
|
|
|
llm_provisioning = LLMProvisioning.PLATFORM
|
|
pricing = Pricing(
|
|
price_per_call_usd=0.0,
|
|
caller_pays_llm=False,
|
|
notes="Deterministic GitHub inspection. Reads only ctx.llm platform grant metadata; never reads provider keys directly.",
|
|
)
|
|
resources = Resources(cpu="1", memory="1Gi", max_runtime_seconds=900)
|
|
egress = EgressPolicy(allow_hosts=("api.github.com",))
|
|
workspace_access = WorkspaceAccess.dynamic(
|
|
max_files=128,
|
|
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
|
|
require_reason=False,
|
|
)
|
|
tools_used = ("github-rest-api", "httpx", "deterministic-static-analysis")
|
|
|
|
@skill(
|
|
description="Check GitHub authentication without logging or returning the token",
|
|
timeout_seconds=30,
|
|
idempotent=True,
|
|
cost_class="cheap",
|
|
)
|
|
async def check_auth(
|
|
self,
|
|
ctx: RunContext[NoAuth],
|
|
github_token: str = "",
|
|
) -> dict[str, Any]:
|
|
_ = ctx.llm # Read the scoped platform grant object without exposing it.
|
|
token = _resolve_github_token(ctx, github_token)
|
|
if not token:
|
|
return {
|
|
"ok": False,
|
|
"authenticated": False,
|
|
"message": "Provide github_token or configure a consumer secret named github_token. Required scopes: repo/read for private repository inspection; issues:write or repo for issue creation.",
|
|
}
|
|
async with _github_client(self.config.github_api_base, token) as client:
|
|
result = await _github_get(client, "/user")
|
|
if not result["ok"]:
|
|
return _error_result("github_auth_failed", result)
|
|
data = result["json"]
|
|
return {
|
|
"ok": True,
|
|
"authenticated": True,
|
|
"login": data.get("login"),
|
|
"id": data.get("id"),
|
|
"scopes": result.get("scopes", []),
|
|
"rate_limit_remaining": result.get("rate_limit_remaining"),
|
|
"message": "Authenticated. Tokens are never stored or returned.",
|
|
}
|
|
|
|
@skill(
|
|
description="List GitHub repositories accessible to the supplied or configured GitHub token",
|
|
timeout_seconds=60,
|
|
idempotent=True,
|
|
cost_class="cheap",
|
|
)
|
|
async def list_repositories(
|
|
self,
|
|
ctx: RunContext[NoAuth],
|
|
github_token: str = "",
|
|
visibility: str = "all",
|
|
affiliation: str = "owner,collaborator,organization_member",
|
|
per_page: int = 100,
|
|
max_pages: int = 10,
|
|
) -> dict[str, Any]:
|
|
_ = ctx.llm
|
|
await ctx.emit_progress("auth")
|
|
token = _resolve_github_token(ctx, github_token)
|
|
if not token:
|
|
return _missing_token_result()
|
|
per_page = max(1, min(int(per_page), 100))
|
|
max_pages = max(1, min(int(max_pages), 50))
|
|
repos: list[dict[str, Any]] = []
|
|
async with _github_client(self.config.github_api_base, token) as client:
|
|
for page in range(1, max_pages + 1):
|
|
result = await _github_get(
|
|
client,
|
|
"/user/repos",
|
|
params={
|
|
"visibility": visibility,
|
|
"affiliation": affiliation,
|
|
"sort": "full_name",
|
|
"direction": "asc",
|
|
"per_page": per_page,
|
|
"page": page,
|
|
},
|
|
)
|
|
if not result["ok"]:
|
|
return _error_result("github_repo_list_failed", result)
|
|
batch = result["json"] or []
|
|
for repo in batch:
|
|
repos.append(
|
|
{
|
|
"owner": repo.get("owner", {}).get("login"),
|
|
"name": repo.get("name"),
|
|
"full_name": repo.get("full_name"),
|
|
"default_branch": repo.get("default_branch"),
|
|
"private": bool(repo.get("private")),
|
|
"archived": bool(repo.get("archived")),
|
|
"fork": bool(repo.get("fork")),
|
|
"html_url": repo.get("html_url"),
|
|
"description": repo.get("description"),
|
|
"language": repo.get("language"),
|
|
"updated_at": repo.get("updated_at"),
|
|
}
|
|
)
|
|
await ctx.emit_progress(f"repo list page {page}: {len(repos)} repos")
|
|
if len(batch) < per_page:
|
|
break
|
|
return {"ok": True, "repositories": repos, "count": len(repos)}
|
|
|
|
@skill(
|
|
description="Inspect a complete GitHub repository tree and return structured JSON plus markdown report with file-level citations",
|
|
timeout_seconds=900,
|
|
idempotent=True,
|
|
cost_class="analysis",
|
|
)
|
|
async def inspect_repository(
|
|
self,
|
|
ctx: RunContext[NoAuth],
|
|
owner: str,
|
|
repo: str,
|
|
github_token: str = "",
|
|
ref: str = "",
|
|
include_hidden: bool = True,
|
|
max_files: int = 750,
|
|
max_bytes: int = 4_000_000,
|
|
max_file_bytes: int = 250_000,
|
|
) -> dict[str, Any]:
|
|
creds = ctx.llm
|
|
await ctx.emit_progress("auth")
|
|
token = _resolve_github_token(ctx, github_token)
|
|
if not token:
|
|
return _missing_token_result()
|
|
|
|
safe_owner = _clean_repo_part(owner)
|
|
safe_repo = _clean_repo_part(repo)
|
|
if not safe_owner or not safe_repo:
|
|
return {"ok": False, "error": {"code": "invalid_repo", "message": "owner and repo are required"}}
|
|
|
|
max_files = max(1, min(int(max_files), 5000))
|
|
max_bytes = max(10_000, min(int(max_bytes), 25_000_000))
|
|
max_file_bytes = max(1_000, min(int(max_file_bytes), 2_000_000))
|
|
|
|
async with _github_client(self.config.github_api_base, token) as client:
|
|
repo_result = await _github_get(client, f"/repos/{safe_owner}/{safe_repo}")
|
|
if not repo_result["ok"]:
|
|
return _error_result("github_repo_metadata_failed", repo_result)
|
|
repo_meta = repo_result["json"]
|
|
target_ref = ref.strip() or repo_meta.get("default_branch") or "HEAD"
|
|
|
|
await ctx.emit_progress("branch/ref")
|
|
branch_result = await _github_get(client, f"/repos/{safe_owner}/{safe_repo}/branches")
|
|
branches = []
|
|
if branch_result["ok"]:
|
|
branches = [
|
|
{"name": item.get("name"), "protected": item.get("protected", False)}
|
|
for item in (branch_result["json"] or [])
|
|
]
|
|
|
|
await ctx.emit_progress("tree fetch")
|
|
tree_result = await _github_get(
|
|
client,
|
|
f"/repos/{safe_owner}/{safe_repo}/git/trees/{target_ref}",
|
|
params={"recursive": "1"},
|
|
)
|
|
if not tree_result["ok"]:
|
|
return _error_result("github_tree_fetch_failed", tree_result)
|
|
tree_json = tree_result["json"] or {}
|
|
tree_entries = tree_json.get("tree") or []
|
|
if not tree_entries:
|
|
return {
|
|
"ok": True,
|
|
"repository": _repo_payload(repo_meta, target_ref, branches),
|
|
"report": {
|
|
"markdown": f"# Repository inspection: {safe_owner}/{safe_repo}\n\nThe repository tree is empty for ref `{target_ref}`.",
|
|
"json": {},
|
|
},
|
|
"warnings": ["Repository tree was empty for the selected ref."],
|
|
}
|
|
|
|
await ctx.emit_progress("classification")
|
|
classified, skipped, candidate_files = _classify_tree(tree_entries, include_hidden)
|
|
limited_candidates = []
|
|
total_candidate_bytes = 0
|
|
for item in candidate_files:
|
|
if len(limited_candidates) >= max_files:
|
|
skipped.append(_skip_record(item, "max_files_limit", f"max_files={max_files} reached"))
|
|
continue
|
|
size = int(item.get("size") or 0)
|
|
if size > max_file_bytes:
|
|
skipped.append(_skip_record(item, "too_large", f"file size {size} exceeds max_file_bytes={max_file_bytes}"))
|
|
continue
|
|
if total_candidate_bytes + size > max_bytes:
|
|
skipped.append(_skip_record(item, "max_bytes_limit", f"aggregate text byte budget max_bytes={max_bytes} reached"))
|
|
continue
|
|
limited_candidates.append(item)
|
|
total_candidate_bytes += size
|
|
|
|
await ctx.emit_progress(f"file fetch: {len(limited_candidates)} text files")
|
|
files: list[RepoFile] = []
|
|
for index, item in enumerate(limited_candidates, start=1):
|
|
await ctx.check_cancelled()
|
|
if index == 1 or index % 25 == 0:
|
|
await ctx.emit_progress(f"file fetch {index}/{len(limited_candidates)}")
|
|
blob = await _github_get(client, f"/repos/{safe_owner}/{safe_repo}/git/blobs/{item['sha']}")
|
|
if not blob["ok"]:
|
|
skipped.append(_skip_record(item, "blob_fetch_failed", _github_error_message(blob)))
|
|
continue
|
|
repo_file = _decode_blob(item, blob["json"] or {})
|
|
if repo_file is None:
|
|
skipped.append(_skip_record(item, "decode_failed", "GitHub blob could not be decoded as supported text"))
|
|
continue
|
|
if _looks_binary(repo_file.text):
|
|
skipped.append(_skip_record(item, "binary_content", "NUL/control-byte heuristic indicates binary content"))
|
|
continue
|
|
files.append(repo_file)
|
|
|
|
await ctx.emit_progress("analysis")
|
|
analysis = _analyze_repository(
|
|
repo_meta=repo_meta,
|
|
ref=target_ref,
|
|
branches=branches,
|
|
tree_entries=tree_entries,
|
|
classified=classified,
|
|
files=files,
|
|
skipped=skipped,
|
|
limits={"max_files": max_files, "max_bytes": max_bytes, "max_file_bytes": max_file_bytes},
|
|
llm_creds=creds,
|
|
)
|
|
markdown = _render_markdown(analysis)
|
|
await ctx.emit_progress("report")
|
|
artifact = await ctx.write_artifact(
|
|
f"github-repo-inspection-{safe_owner}-{safe_repo}.md",
|
|
markdown.encode("utf-8"),
|
|
"text/markdown",
|
|
)
|
|
await ctx.emit_artifact(artifact)
|
|
analysis["artifacts"] = [{"name": artifact.name, "uri": artifact.uri, "mime_type": artifact.mime_type}]
|
|
return {"ok": True, "report": {"json": analysis, "markdown": markdown}, **analysis}
|
|
|
|
@skill(
|
|
description="Create a GitHub issue for a confirmed repository inspection recommendation or risk",
|
|
timeout_seconds=60,
|
|
cost_class="write",
|
|
)
|
|
async def create_issue(
|
|
self,
|
|
ctx: RunContext[NoAuth],
|
|
owner: str,
|
|
repo: str,
|
|
title: str,
|
|
body: str,
|
|
github_token: str = "",
|
|
labels: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
_ = ctx.llm
|
|
token = _resolve_github_token(ctx, github_token)
|
|
if not token:
|
|
return _missing_token_result(action="create issues")
|
|
safe_owner = _clean_repo_part(owner)
|
|
safe_repo = _clean_repo_part(repo)
|
|
clean_title = str(title or "").strip()[:256]
|
|
clean_body = _redact_secrets(str(body or ""))[:60_000]
|
|
if not clean_title or not clean_body:
|
|
return {"ok": False, "error": {"code": "invalid_issue", "message": "title and body are required"}}
|
|
payload: dict[str, Any] = {"title": clean_title, "body": clean_body}
|
|
if labels:
|
|
payload["labels"] = [str(label)[:50] for label in labels[:10]]
|
|
async with _github_client(self.config.github_api_base, token) as client:
|
|
result = await _github_post(client, f"/repos/{safe_owner}/{safe_repo}/issues", json_payload=payload)
|
|
if not result["ok"]:
|
|
return _error_result("github_issue_create_failed", result)
|
|
issue = result["json"]
|
|
return {
|
|
"ok": True,
|
|
"issue": {
|
|
"number": issue.get("number"),
|
|
"title": issue.get("title"),
|
|
"html_url": issue.get("html_url"),
|
|
"state": issue.get("state"),
|
|
},
|
|
}
|
|
|
|
def _build_deep_agent(self, *, ctx: RunContext[NoAuth], creds: LLMCreds) -> Any:
|
|
"""Optional inner DeepAgent builder for future narrative enhancement.
|
|
|
|
The public scanner is deterministic to guarantee full-tree coverage;
|
|
this builder wires packaged DeepAgents skills correctly without using
|
|
provider keys directly.
|
|
"""
|
|
if not creds.api_key:
|
|
raise RuntimeError("LLM credentials were not available from ctx.llm.")
|
|
from deepagents import create_deep_agent
|
|
from langchain_openai import ChatOpenAI
|
|
|
|
model = ChatOpenAI(model=creds.model, base_url=creds.base_url, api_key=creds.api_key, temperature=0)
|
|
backend = ctx.workspace_backend()
|
|
skill_sources = _seed_runtime_skills(backend, ctx)
|
|
return create_deep_agent(
|
|
model=model,
|
|
backend=backend,
|
|
skills=skill_sources or None,
|
|
system_prompt=(
|
|
"You review repository inspection JSON and improve clarity without "
|
|
"inventing findings. Preserve file-level citations."
|
|
),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class RepoFile:
|
|
path: str
|
|
sha: str
|
|
size: int
|
|
category: str
|
|
text: str
|
|
lines: list[str]
|
|
|
|
|
|
def _github_client(api_base: str, token: str) -> httpx.AsyncClient:
|
|
return httpx.AsyncClient(
|
|
base_url=api_base.rstrip("/"),
|
|
timeout=httpx.Timeout(30.0, connect=10.0),
|
|
headers={
|
|
"Authorization": f"Bearer {token}",
|
|
"Accept": "application/vnd.github+json",
|
|
"X-GitHub-Api-Version": "2022-11-28",
|
|
"User-Agent": "a2a-github-repo-inspector/1.0",
|
|
},
|
|
follow_redirects=True,
|
|
)
|
|
|
|
|
|
async def _github_get(client: httpx.AsyncClient, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
try:
|
|
resp = await client.get(path, params=params)
|
|
return _response_payload(resp)
|
|
except httpx.HTTPError as exc:
|
|
return {"ok": False, "status_code": 0, "message": str(exc), "json": None}
|
|
|
|
|
|
async def _github_post(client: httpx.AsyncClient, path: str, json_payload: dict[str, Any]) -> dict[str, Any]:
|
|
try:
|
|
resp = await client.post(path, json=json_payload)
|
|
return _response_payload(resp)
|
|
except httpx.HTTPError as exc:
|
|
return {"ok": False, "status_code": 0, "message": str(exc), "json": None}
|
|
|
|
|
|
def _response_payload(resp: httpx.Response) -> dict[str, Any]:
|
|
try:
|
|
data = resp.json() if resp.content else None
|
|
except ValueError:
|
|
data = {"message": resp.text[:500]}
|
|
scopes = [s.strip() for s in resp.headers.get("x-oauth-scopes", "").split(",") if s.strip()]
|
|
payload = {
|
|
"ok": 200 <= resp.status_code < 300,
|
|
"status_code": resp.status_code,
|
|
"json": data,
|
|
"message": (data or {}).get("message") if isinstance(data, dict) else resp.text[:500],
|
|
"rate_limit_remaining": resp.headers.get("x-ratelimit-remaining"),
|
|
"rate_limit_reset": resp.headers.get("x-ratelimit-reset"),
|
|
"scopes": scopes,
|
|
}
|
|
return payload
|
|
|
|
|
|
def _resolve_github_token(ctx: RunContext[NoAuth], github_token: str = "") -> str:
|
|
token = str(github_token or "").strip()
|
|
if token:
|
|
return token
|
|
for name in ("github_token", "GITHUB_TOKEN"):
|
|
try:
|
|
token = ctx.consumer_secret(name).strip()
|
|
if token:
|
|
return token
|
|
except Exception:
|
|
pass
|
|
return ""
|
|
|
|
|
|
def _missing_token_result(action: str = "inspect repositories") -> dict[str, Any]:
|
|
return {
|
|
"ok": False,
|
|
"error": {
|
|
"code": "missing_github_token",
|
|
"message": f"Provide github_token or configure a consumer secret named github_token to {action}.",
|
|
"required_scopes": ["repo/read for private repositories", "issues:write or repo for issue creation"],
|
|
},
|
|
}
|
|
|
|
|
|
def _error_result(code: str, result: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"ok": False,
|
|
"error": {
|
|
"code": code,
|
|
"status_code": result.get("status_code"),
|
|
"message": _github_error_message(result),
|
|
"rate_limit_remaining": result.get("rate_limit_remaining"),
|
|
"rate_limit_reset": result.get("rate_limit_reset"),
|
|
},
|
|
}
|
|
|
|
|
|
def _github_error_message(result: dict[str, Any]) -> str:
|
|
status = result.get("status_code")
|
|
msg = str(result.get("message") or "GitHub API request failed")
|
|
if status == 401:
|
|
return "GitHub authentication failed. Check token validity and scopes."
|
|
if status == 403 and result.get("rate_limit_remaining") == "0":
|
|
return f"GitHub rate limit exceeded. Reset epoch: {result.get('rate_limit_reset')}"
|
|
if status == 404:
|
|
return "Repository, ref, or resource was not found, or token lacks access."
|
|
return msg[:500]
|
|
|
|
|
|
def _clean_repo_part(value: str) -> str:
|
|
return re.sub(r"[^A-Za-z0-9_.-]", "", str(value or "").strip())[:100]
|
|
|
|
|
|
def _repo_payload(meta: dict[str, Any], ref: str, branches: list[dict[str, Any]]) -> dict[str, Any]:
|
|
return {
|
|
"owner": meta.get("owner", {}).get("login"),
|
|
"name": meta.get("name"),
|
|
"full_name": meta.get("full_name"),
|
|
"default_branch": meta.get("default_branch"),
|
|
"selected_ref": ref,
|
|
"private": bool(meta.get("private")),
|
|
"html_url": meta.get("html_url"),
|
|
"description": meta.get("description"),
|
|
"language": meta.get("language"),
|
|
"branches": branches,
|
|
}
|
|
|
|
|
|
def _classify_tree(tree_entries: list[dict[str, Any]], include_hidden: bool) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]:
|
|
categories: dict[str, list[dict[str, Any]]] = {cat: [] for cat in CATEGORY_ORDER}
|
|
skipped: list[dict[str, Any]] = []
|
|
candidate_files: list[dict[str, Any]] = []
|
|
for item in sorted(tree_entries, key=lambda x: x.get("path", "")):
|
|
path = item.get("path", "")
|
|
if not path:
|
|
continue
|
|
category = classify_path(path)
|
|
entry = {
|
|
"path": path,
|
|
"type": item.get("type"),
|
|
"mode": item.get("mode"),
|
|
"size": item.get("size"),
|
|
"sha": item.get("sha"),
|
|
"category": category,
|
|
"hidden": _is_hidden_path(path),
|
|
}
|
|
categories.setdefault(category, []).append(entry)
|
|
if item.get("type") != "blob":
|
|
continue
|
|
if _is_hidden_path(path) and not include_hidden:
|
|
skipped.append(_skip_record(entry, "hidden_excluded", "include_hidden=false"))
|
|
continue
|
|
skip_reason = _skip_reason_for_path(path)
|
|
if skip_reason:
|
|
skipped.append(_skip_record(entry, skip_reason[0], skip_reason[1]))
|
|
continue
|
|
candidate_files.append({**item, "category": category})
|
|
return {
|
|
"categories": categories,
|
|
"tree": _tree_from_paths([item.get("path", "") for item in tree_entries if item.get("path")]),
|
|
"counts_by_category": {k: len(v) for k, v in categories.items()},
|
|
"total_entries": len(tree_entries),
|
|
"total_files": sum(1 for item in tree_entries if item.get("type") == "blob"),
|
|
}, skipped, candidate_files
|
|
|
|
|
|
def classify_path(path: str) -> str:
|
|
p = path.replace("\\", "/")
|
|
lower = p.lower()
|
|
name = lower.rsplit("/", 1)[-1]
|
|
parts = lower.split("/")
|
|
ext = Path(name).suffix.lower()
|
|
if _is_hidden_path(p):
|
|
if name in {".github", ".gitlab-ci.yml", ".circleci"} or ".github/workflows/" in lower:
|
|
return "CI/CD"
|
|
if name in {".dockerignore"}:
|
|
return "Docker/container"
|
|
if name.startswith(".env") or name in {".npmrc", ".pypirc", ".netrc", ".gitignore", ".gitattributes", ".editorconfig", ".prettierrc", ".eslintrc"}:
|
|
return "hidden/dotfiles"
|
|
if ".github/workflows/" in lower or name in {"jenkinsfile", ".gitlab-ci.yml", "circle.yml", "azure-pipelines.yml", "bitbucket-pipelines.yml", "buildkite.yml"} or ".circleci/" in lower:
|
|
return "CI/CD"
|
|
if name == "dockerfile" or name.startswith("dockerfile") or name in {"docker-compose.yml", "docker-compose.yaml", "compose.yml", "compose.yaml", ".dockerignore"}:
|
|
return "Docker/container"
|
|
if any(part in {"test", "tests", "spec", "specs", "__tests__", "e2e"} for part in parts) or re.search(r"(test|spec)\.(py|js|jsx|ts|tsx|go|rs|rb|php|java)$", lower):
|
|
return "tests"
|
|
if name in {"readme.md", "readme.rst", "readme.txt", "license", "license.md", "changelog.md", "contributing.md", "code_of_conduct.md", "security.md"} or parts[0] in {"docs", "doc", "documentation", "examples"} or ext in {".md", ".rst"}:
|
|
return "docs"
|
|
if name in {"package.json", "package-lock.json", "yarn.lock", "pnpm-lock.yaml", "requirements.txt", "poetry.lock", "pyproject.toml", "pipfile", "pipfile.lock", "go.mod", "go.sum", "cargo.toml", "cargo.lock", "pom.xml", "build.gradle", "gradle.lockfile", "gemfile", "gemfile.lock", "composer.json", "composer.lock", "mix.exs", "deno.lock"}:
|
|
return "package/dependency"
|
|
if ext in {".tf", ".tfvars", ".hcl"} or any(part in {"terraform", "infra", "infrastructure", "k8s", "kubernetes", "helm", "charts", "ansible", "pulumi", "cloudformation"} for part in parts) or name in {"serverless.yml", "serverless.yaml", "template.yaml", "template.yml"}:
|
|
return "infra/IaC"
|
|
if parts[0] in {"scripts", "bin", "tools"} or ext in {".sh", ".bash", ".zsh", ".ps1", ".bat", ".cmd"}:
|
|
return "scripts"
|
|
if ext in {".yml", ".yaml", ".json", ".toml", ".ini", ".cfg", ".conf", ".xml", ".properties"}:
|
|
return "config"
|
|
if ext in {".py", ".js", ".jsx", ".ts", ".tsx", ".go", ".rs", ".java", ".rb", ".php", ".cs", ".cpp", ".c", ".h", ".swift", ".kt", ".scala", ".sql", ".graphql", ".proto", ".html", ".css", ".scss"}:
|
|
return "source"
|
|
return "assets/other"
|
|
|
|
|
|
def _is_hidden_path(path: str) -> bool:
|
|
return any(part.startswith(".") and part not in {".", ".."} for part in path.split("/"))
|
|
|
|
|
|
def _skip_reason_for_path(path: str) -> tuple[str, str] | None:
|
|
parts = path.replace("\\", "/").split("/")
|
|
for part in parts[:-1]:
|
|
if part.lower() in VENDOR_BUILD_CACHE_DIRS:
|
|
return ("vendor_build_cache", f"Skipped folder '{part}' as vendor/build/cache output")
|
|
ext = Path(path).suffix.lower()
|
|
if ext in BINARY_EXTENSIONS:
|
|
return ("binary_extension", f"Skipped binary/asset extension {ext}")
|
|
if ext and ext not in TEXT_EXTENSIONS and classify_path(path) == "assets/other":
|
|
guessed, _ = mimetypes.guess_type(path)
|
|
if guessed and not guessed.startswith("text/"):
|
|
return ("unsupported_asset", f"Skipped unsupported asset MIME guess {guessed}")
|
|
return None
|
|
|
|
|
|
def _skip_record(item: dict[str, Any], reason: str, detail: str) -> dict[str, Any]:
|
|
return {
|
|
"path": item.get("path"),
|
|
"sha": item.get("sha"),
|
|
"size": item.get("size"),
|
|
"category": item.get("category") or classify_path(str(item.get("path") or "")),
|
|
"reason": reason,
|
|
"detail": detail,
|
|
}
|
|
|
|
|
|
def _tree_from_paths(paths: list[str]) -> list[dict[str, Any]]:
|
|
# Compact flat tree is easier for UI filters while still preserving full paths.
|
|
return [{"path": path, "depth": path.count("/"), "name": path.rsplit("/", 1)[-1], "hidden": _is_hidden_path(path), "category": classify_path(path)} for path in sorted(paths)]
|
|
|
|
|
|
def _decode_blob(item: dict[str, Any], blob_json: dict[str, Any]) -> RepoFile | None:
|
|
content = blob_json.get("content") or ""
|
|
encoding = blob_json.get("encoding")
|
|
if encoding != "base64":
|
|
return None
|
|
try:
|
|
raw = base64.b64decode(content, validate=False)
|
|
except Exception:
|
|
return None
|
|
try:
|
|
text = raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
try:
|
|
text = raw.decode("latin-1")
|
|
except UnicodeDecodeError:
|
|
return None
|
|
text = text.replace("\r\n", "\n").replace("\r", "\n")
|
|
return RepoFile(
|
|
path=item.get("path", ""),
|
|
sha=item.get("sha", ""),
|
|
size=int(item.get("size") or len(raw)),
|
|
category=item.get("category") or classify_path(item.get("path", "")),
|
|
text=text,
|
|
lines=text.split("\n"),
|
|
)
|
|
|
|
|
|
def _looks_binary(text: str) -> bool:
|
|
if "\x00" in text:
|
|
return True
|
|
if not text:
|
|
return False
|
|
sample = text[:4096]
|
|
control = sum(1 for ch in sample if ord(ch) < 9 or (13 < ord(ch) < 32))
|
|
return control / max(len(sample), 1) > 0.05
|
|
|
|
|
|
def _analyze_repository(
|
|
*,
|
|
repo_meta: dict[str, Any],
|
|
ref: str,
|
|
branches: list[dict[str, Any]],
|
|
tree_entries: list[dict[str, Any]],
|
|
classified: dict[str, Any],
|
|
files: list[RepoFile],
|
|
skipped: list[dict[str, Any]],
|
|
limits: dict[str, int],
|
|
llm_creds: LLMCreds,
|
|
) -> dict[str, Any]:
|
|
findings: list[dict[str, Any]] = []
|
|
recommendations: list[dict[str, Any]] = []
|
|
file_index = {f.path: f for f in files}
|
|
file_paths = {item.get("path", "") for item in tree_entries}
|
|
categories = classified["counts_by_category"]
|
|
|
|
language_counter = _infer_languages(file_paths)
|
|
frameworks = _infer_frameworks(file_index, file_paths)
|
|
entrypoints = _infer_entrypoints(file_paths, file_index)
|
|
external_services = _infer_external_services(files)
|
|
|
|
for repo_file in files:
|
|
findings.extend(_security_findings(repo_file))
|
|
findings.extend(_todo_findings(repo_file))
|
|
findings.extend(_commented_code_findings(repo_file))
|
|
findings.extend(_docker_findings(repo_file))
|
|
findings.extend(_ci_findings(repo_file))
|
|
findings.extend(_iac_findings(repo_file))
|
|
findings.extend(_dependency_findings(repo_file))
|
|
findings.extend(_config_findings(repo_file))
|
|
|
|
structural_findings, structural_recs = _structural_findings_and_recs(file_paths, file_index, categories)
|
|
findings.extend(structural_findings)
|
|
recommendations.extend(structural_recs)
|
|
recommendations.extend(_recommendations_from_findings(findings))
|
|
|
|
scores = _score(findings, file_paths, categories, skipped)
|
|
architecture = {
|
|
"languages": language_counter,
|
|
"frameworks": frameworks,
|
|
"entrypoints": entrypoints,
|
|
"major_components": _major_components(file_paths),
|
|
"data_flow": _infer_data_flow(files, external_services),
|
|
"build_test_deploy_path": _infer_build_test_deploy(file_paths, file_index),
|
|
"external_services": external_services,
|
|
}
|
|
|
|
skipped_by_reason = Counter(item["reason"] for item in skipped)
|
|
analysis = {
|
|
"repository": _repo_payload(repo_meta, ref, branches),
|
|
"inspection_metadata": {
|
|
"tree_entries_seen": len(tree_entries),
|
|
"files_seen": sum(1 for item in tree_entries if item.get("type") == "blob"),
|
|
"text_files_analyzed": len(files),
|
|
"bytes_analyzed": sum(f.size for f in files),
|
|
"limits": limits,
|
|
"llm_grant_source": getattr(llm_creds, "source", "platform"),
|
|
"note": "Findings are deterministic heuristics and evidence-backed; no secrets are stored or returned.",
|
|
},
|
|
"repo_map": classified,
|
|
"skipped_files": skipped,
|
|
"skipped_summary": dict(skipped_by_reason),
|
|
"scorecard": scores,
|
|
"architecture_summary": architecture,
|
|
"findings": findings,
|
|
"recommendations": recommendations,
|
|
"issue_ready_payloads": [rec.get("issue_payload") for rec in recommendations if rec.get("issue_payload")],
|
|
}
|
|
return analysis
|
|
|
|
|
|
def _citation(repo_file: RepoFile, line_start: int, line_end: int | None = None) -> dict[str, Any]:
|
|
line_end = line_start if line_end is None else line_end
|
|
start = max(line_start, 1)
|
|
end = min(max(line_end, start), len(repo_file.lines))
|
|
snippet = "\n".join(repo_file.lines[start - 1:end])
|
|
return {
|
|
"path": repo_file.path,
|
|
"line_start": start,
|
|
"line_end": end,
|
|
"snippet": _redact_secrets(snippet)[:1200],
|
|
}
|
|
|
|
|
|
def _structure_citation(path: str, note: str) -> dict[str, Any]:
|
|
return {"path": path, "line_start": None, "line_end": None, "snippet": note, "evidence_type": "repository-structure"}
|
|
|
|
|
|
def _finding(category: str, title: str, severity: str, description: str, citation: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"id": _slug(f"{category}-{title}-{citation.get('path')}-{citation.get('line_start')}"),
|
|
"category": category,
|
|
"title": title,
|
|
"severity": severity,
|
|
"description": description,
|
|
"citations": [citation],
|
|
}
|
|
|
|
|
|
def _security_findings(repo_file: RepoFile) -> list[dict[str, Any]]:
|
|
out = []
|
|
for i, line in enumerate(repo_file.lines, start=1):
|
|
for label, pattern in SECRET_REGEXES:
|
|
if pattern.search(line):
|
|
out.append(_finding("security/config", f"Possible hard-coded {label}", "high", "A secret-like value appears in source/config. Verify whether it is real and rotate if exposed.", _citation(repo_file, i)))
|
|
lower = line.lower()
|
|
if "cors" in lower and ("*" in line or "origin: true" in lower):
|
|
out.append(_finding("security/config", "Permissive CORS configuration", "medium", "CORS appears broadly permissive. Confirm this is intentional and constrained in production.", _citation(repo_file, i)))
|
|
if "csrf" in lower and ("false" in lower or "disable" in lower):
|
|
out.append(_finding("security/config", "CSRF protection disabled", "medium", "CSRF appears disabled in configuration or code.", _citation(repo_file, i)))
|
|
if "verify" in lower and "false" in lower and ("ssl" in lower or "tls" in lower or "cert" in lower):
|
|
out.append(_finding("security/config", "TLS/certificate verification disabled", "high", "TLS or certificate verification appears disabled.", _citation(repo_file, i)))
|
|
if re.search(r"0\.0\.0\.0|host\s*[:=]\s*['\"]?\*", line, re.I) and repo_file.category in {"config", "source", "Docker/container"}:
|
|
out.append(_finding("security/config", "Broad network binding", "low", "Service binds broadly; verify production network exposure and firewalling.", _citation(repo_file, i)))
|
|
if re.search(r"chmod\s+777|--privileged|privileged\s*[:=]\s*true", line, re.I):
|
|
out.append(_finding("security/config", "Overly broad permissions", "high", "Command or configuration grants broad filesystem/container permissions.", _citation(repo_file, i)))
|
|
if re.search(r"eval\(|exec\(|child_process|shell=True|pickle\.loads|yaml\.load\(", line):
|
|
out.append(_finding("security/config", "Dangerous execution/deserialization primitive", "medium", "Potentially dangerous execution or deserialization API appears in code. Review input trust boundaries.", _citation(repo_file, i)))
|
|
return out
|
|
|
|
|
|
def _todo_findings(repo_file: RepoFile) -> list[dict[str, Any]]:
|
|
return [_finding("dead-code", "TODO/FIXME/HACK marker", "low", "Maintenance marker found; verify whether it represents unfinished production work.", _citation(repo_file, i)) for i, line in enumerate(repo_file.lines, start=1) if TODO_RE.search(line)]
|
|
|
|
|
|
def _commented_code_findings(repo_file: RepoFile) -> list[dict[str, Any]]:
|
|
out = []
|
|
for i, line in enumerate(repo_file.lines, start=1):
|
|
if COMMENTED_BLOCK_RE.search(line):
|
|
out.append(_finding("dead-code", "Commented-out code heuristic", "low", "Line looks like commented-out code. Remove or document if intentionally retained.", _citation(repo_file, i)))
|
|
return out[:20]
|
|
|
|
|
|
def _docker_findings(repo_file: RepoFile) -> list[dict[str, Any]]:
|
|
if repo_file.category != "Docker/container":
|
|
return []
|
|
out = []
|
|
has_user = any(re.match(r"\s*USER\s+", line, re.I) for line in repo_file.lines)
|
|
for i, line in enumerate(repo_file.lines, start=1):
|
|
if re.search(r"FROM\s+[^\s:]+\s*$", line, re.I) or ":latest" in line:
|
|
out.append(_finding("ci-docker-infra", "Unpinned or latest Docker base image", "medium", "Docker base image appears unpinned or uses latest, reducing reproducibility.", _citation(repo_file, i)))
|
|
if re.search(r"curl .*\|\s*(sh|bash)|wget .*\|\s*(sh|bash)", line, re.I):
|
|
out.append(_finding("ci-docker-infra", "Remote script piped to shell", "high", "Docker build pipes remote content into a shell. Pin and verify downloaded artifacts.", _citation(repo_file, i)))
|
|
if "--no-check-certificate" in line or "strict-ssl false" in line.lower():
|
|
out.append(_finding("ci-docker-infra", "TLS checks disabled in container build", "high", "Container build disables TLS verification.", _citation(repo_file, i)))
|
|
if repo_file.path.lower().endswith("dockerfile") and not has_user:
|
|
out.append(_finding("ci-docker-infra", "Dockerfile lacks non-root USER", "medium", "No USER directive was found; container may run as root by default.", _structure_citation(repo_file.path, "Dockerfile inspected; no USER directive found.")))
|
|
return out
|
|
|
|
|
|
def _ci_findings(repo_file: RepoFile) -> list[dict[str, Any]]:
|
|
if repo_file.category != "CI/CD":
|
|
return []
|
|
out = []
|
|
for i, line in enumerate(repo_file.lines, start=1):
|
|
if re.search(r"pull_request_target|GITHUB_TOKEN|permissions:\s*write-all|contents:\s*write", line, re.I):
|
|
out.append(_finding("ci-docker-infra", "Risky CI/CD permission or trigger", "medium", "CI configuration uses a trigger or permission that should be reviewed for least privilege.", _citation(repo_file, i)))
|
|
if re.search(r"curl .*\|\s*(sh|bash)|wget .*\|\s*(sh|bash)", line, re.I):
|
|
out.append(_finding("ci-docker-infra", "Remote CI script piped to shell", "high", "CI downloads and executes remote content without visible verification.", _citation(repo_file, i)))
|
|
return out
|
|
|
|
|
|
def _iac_findings(repo_file: RepoFile) -> list[dict[str, Any]]:
|
|
if repo_file.category != "infra/IaC":
|
|
return []
|
|
out = []
|
|
for i, line in enumerate(repo_file.lines, start=1):
|
|
if re.search(r"0\.0\.0\.0/0|::/0", line):
|
|
out.append(_finding("ci-docker-infra", "Broad network exposure in IaC", "high", "Infrastructure code references world-open CIDR. Confirm security-group/firewall intent.", _citation(repo_file, i)))
|
|
if re.search(r"publicly_accessible\s*=\s*true|public_access\s*=\s*true", line, re.I):
|
|
out.append(_finding("ci-docker-infra", "Publicly accessible infrastructure", "high", "IaC enables public accessibility. Confirm exposure is intentional and protected.", _citation(repo_file, i)))
|
|
return out
|
|
|
|
|
|
def _dependency_findings(repo_file: RepoFile) -> list[dict[str, Any]]:
|
|
if repo_file.category != "package/dependency":
|
|
return []
|
|
out = []
|
|
for i, line in enumerate(repo_file.lines, start=1):
|
|
if re.search(r"[\"']\s*[:=]\s*[\"'][\^~*]|>=|latest", line) and repo_file.path.endswith(("package.json", "pyproject.toml", "requirements.txt", "Gemfile", "composer.json")):
|
|
out.append(_finding("dependency", "Broad or floating dependency version", "low", "Dependency version appears broad/floating. Pin or use lockfiles for reproducible builds.", _citation(repo_file, i)))
|
|
for pkg in RISKY_PACKAGE_HINTS:
|
|
if re.search(rf"[\"']?{re.escape(pkg)}[\"']?", line, re.I):
|
|
out.append(_finding("dependency", f"Known risky package name: {pkg}", "medium", "Package name has a history of incidents or deprecation. Verify necessity and current safety.", _citation(repo_file, i)))
|
|
return out
|
|
|
|
|
|
def _config_findings(repo_file: RepoFile) -> list[dict[str, Any]]:
|
|
out = []
|
|
if repo_file.path.lower().endswith((".env", ".env.local", ".npmrc", ".pypirc", ".netrc")):
|
|
out.append(_finding("security/config", "Sensitive dotfile committed", "medium", "A file commonly used for local secrets/configuration is committed. Verify it contains only placeholders.", _structure_citation(repo_file.path, "Sensitive configuration filename present in repository.")))
|
|
return out
|
|
|
|
|
|
def _structural_findings_and_recs(file_paths: set[str], file_index: dict[str, RepoFile], categories: dict[str, int]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
|
findings = []
|
|
recs = []
|
|
lower_paths = {p.lower() for p in file_paths}
|
|
has_readme = any(Path(p).name.lower().startswith("readme") for p in file_paths)
|
|
has_tests = categories.get("tests", 0) > 0
|
|
has_ci = categories.get("CI/CD", 0) > 0
|
|
has_docker = categories.get("Docker/container", 0) > 0
|
|
has_lock = any(Path(p).name.lower() in {"package-lock.json", "yarn.lock", "pnpm-lock.yaml", "poetry.lock", "pipfile.lock", "go.sum", "cargo.lock", "gemfile.lock", "composer.lock", "gradle.lockfile", "deno.lock"} for p in file_paths)
|
|
has_pkg = categories.get("package/dependency", 0) > 0
|
|
has_security = any(Path(p).name.lower() == "security.md" or p.lower() == ".github/security.md" for p in file_paths)
|
|
has_docs = categories.get("docs", 0) > 0
|
|
root = sorted(p for p in file_paths if "/" not in p)[:1]
|
|
cite_path = root[0] if root else "repository-root"
|
|
if not has_readme:
|
|
findings.append(_finding("docs", "README missing", "medium", "No README file was found, which weakens onboarding and operability.", _structure_citation(cite_path, "No README path found in repository tree.")))
|
|
if not has_tests:
|
|
findings.append(_finding("tests", "Tests missing", "medium", "No test/spec paths were detected. Add automated tests or document external validation.", _structure_citation(cite_path, "No tests/spec paths found in repository tree.")))
|
|
if not has_ci:
|
|
findings.append(_finding("ci-docker-infra", "CI/CD workflow missing", "medium", "No common CI/CD workflow file was detected.", _structure_citation(cite_path, "No CI/CD paths found in repository tree.")))
|
|
if has_pkg and not has_lock:
|
|
findings.append(_finding("dependency", "Dependency lockfile missing", "medium", "Dependency manifests exist but no common lockfile was detected.", _structure_citation(cite_path, "Package/dependency files exist but no lockfile path was detected.")))
|
|
if not has_security:
|
|
recs.append(_recommendation("medium", "Add a SECURITY.md disclosure policy", "Document supported versions and vulnerability reporting channels.", [_structure_citation(cite_path, "No SECURITY.md found.")], "low", "medium", ["security", "documentation"]))
|
|
if not has_docs:
|
|
recs.append(_recommendation("medium", "Add basic repository documentation", "Add README/API docs/runbook/examples covering setup, configuration, testing, and deployment.", [_structure_citation(cite_path, "No docs/readme files found.")], "medium", "high", ["documentation"]))
|
|
if has_docker and ".dockerignore" not in lower_paths:
|
|
recs.append(_recommendation("medium", "Add .dockerignore", "Prevent secrets, build outputs, and dependency folders from entering Docker build context.", [_structure_citation(".dockerignore", "No .dockerignore found while Docker files are present.")], "low", "medium", ["docker", "security"]))
|
|
return findings, recs
|
|
|
|
|
|
def _recommendations_from_findings(findings: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
recs: list[dict[str, Any]] = []
|
|
grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for finding in findings:
|
|
grouped[finding["title"]].append(finding)
|
|
for title, items in grouped.items():
|
|
worst = _worst_severity([i["severity"] for i in items])
|
|
first = items[0]
|
|
citations = [c for item in items[:5] for c in item.get("citations", [])]
|
|
fix = _fix_for_title(title, first["category"])
|
|
recs.append(_recommendation(worst, fix["title"], fix["body"], citations, fix["effort"], fix["impact"], fix["labels"]))
|
|
return recs[:80]
|
|
|
|
|
|
def _fix_for_title(title: str, category: str) -> dict[str, Any]:
|
|
lower = title.lower()
|
|
if "secret" in lower or "token" in lower or "private key" in lower:
|
|
return {"title": "Remove and rotate possible committed secrets", "body": "Move secrets to a secret manager or CI variables, rotate exposed credentials, and add safe examples instead of real values.", "effort": "medium", "impact": "high", "labels": ["security", "secrets"]}
|
|
if "docker" in lower or category == "ci-docker-infra":
|
|
return {"title": f"Harden {title}", "body": "Pin versions, reduce permissions, verify downloads, and document the deployment safety control for the cited configuration.", "effort": "medium", "impact": "high", "labels": ["ci", "docker", "security"]}
|
|
if "dependency" in lower or category == "dependency":
|
|
return {"title": f"Improve dependency hygiene: {title}", "body": "Pin versions where appropriate, commit lockfiles, run audit tooling, and consider SBOM generation in CI.", "effort": "medium", "impact": "medium", "labels": ["dependencies"]}
|
|
if category == "tests":
|
|
return {"title": "Add automated test coverage", "body": "Add unit/integration tests for critical paths and wire them into CI.", "effort": "high", "impact": "high", "labels": ["testing"]}
|
|
if category == "docs":
|
|
return {"title": "Improve repository documentation", "body": "Add setup, configuration, test, deploy, API, and runbook documentation.", "effort": "medium", "impact": "high", "labels": ["documentation"]}
|
|
return {"title": f"Review and remediate: {title}", "body": "Review the cited evidence and either remediate the risk or document why it is acceptable.", "effort": "medium", "impact": "medium", "labels": [category.split("/")[0]]}
|
|
|
|
|
|
def _recommendation(severity: str, title: str, body: str, citations: list[dict[str, Any]], effort: str, impact: str, labels: list[str]) -> dict[str, Any]:
|
|
rec_id = _slug(f"rec-{title}-{citations[0].get('path') if citations else 'repo'}")
|
|
citation_lines = "\n".join(_format_citation_line(c) for c in citations[:10])
|
|
issue_body = _redact_secrets(f"## Recommendation\n{body}\n\n## Evidence\n{citation_lines}\n\n## Suggested fix\n- Apply the remediation above.\n- Add or update tests/docs/CI checks as appropriate.\n- Re-run GitHub Repo Inspector to verify.\n")
|
|
return {
|
|
"id": rec_id,
|
|
"title": title,
|
|
"severity": severity,
|
|
"impact": impact,
|
|
"effort": effort,
|
|
"description": body,
|
|
"citations": citations,
|
|
"issue_payload": {"title": title[:256], "body": issue_body, "labels": labels[:10]},
|
|
}
|
|
|
|
|
|
def _format_citation_line(c: dict[str, Any]) -> str:
|
|
if c.get("line_start"):
|
|
return f"- `{c.get('path')}:{c.get('line_start')}-{c.get('line_end')}`\n ```\n {c.get('snippet', '')[:500]}\n ```"
|
|
return f"- `{c.get('path')}` — {c.get('snippet', '')[:300]}"
|
|
|
|
|
|
def _score(findings: list[dict[str, Any]], file_paths: set[str], categories: dict[str, int], skipped: list[dict[str, Any]]) -> dict[str, Any]:
|
|
buckets = {
|
|
"security": 100,
|
|
"maintainability": 100,
|
|
"testing": 100,
|
|
"docs": 100,
|
|
"CI/CD": 100,
|
|
"deployability": 100,
|
|
"dependency_hygiene": 100,
|
|
"observability_configuration": 100,
|
|
}
|
|
penalties = {"critical": 25, "high": 16, "medium": 9, "low": 3}
|
|
for f in findings:
|
|
sev = f.get("severity", "low")
|
|
p = penalties.get(sev, 3)
|
|
cat = f.get("category")
|
|
if cat == "security/config":
|
|
buckets["security"] -= p
|
|
buckets["observability_configuration"] -= max(2, p // 3)
|
|
elif cat == "dependency":
|
|
buckets["dependency_hygiene"] -= p
|
|
elif cat == "tests":
|
|
buckets["testing"] -= p
|
|
elif cat == "docs":
|
|
buckets["docs"] -= p
|
|
elif cat == "ci-docker-infra":
|
|
buckets["CI/CD"] -= p
|
|
buckets["deployability"] -= max(2, p // 2)
|
|
elif cat == "dead-code":
|
|
buckets["maintainability"] -= min(p, 2)
|
|
if categories.get("tests", 0) == 0:
|
|
buckets["testing"] -= 20
|
|
if categories.get("docs", 0) == 0:
|
|
buckets["docs"] -= 20
|
|
if categories.get("CI/CD", 0) == 0:
|
|
buckets["CI/CD"] -= 20
|
|
if not any(Path(p).name.lower() in {"dockerfile", "docker-compose.yml", "compose.yml", "serverless.yml", "template.yaml"} or p.endswith(".tf") for p in file_paths):
|
|
buckets["deployability"] -= 8
|
|
if skipped:
|
|
buckets["maintainability"] -= min(10, len(skipped) // 20)
|
|
sub_scores = {k: max(0, min(100, int(v))) for k, v in buckets.items()}
|
|
total = int(round(sum(sub_scores.values()) / len(sub_scores)))
|
|
return {
|
|
"total": total,
|
|
"sub_scores": sub_scores,
|
|
"drivers": [f"{len(findings)} evidence-backed findings", f"{categories.get('tests', 0)} test files/entries", f"{categories.get('docs', 0)} docs files/entries", f"{categories.get('CI/CD', 0)} CI/CD entries", f"{len(skipped)} skipped/limited files explicitly reported"],
|
|
}
|
|
|
|
|
|
def _infer_languages(file_paths: set[str]) -> dict[str, int]:
|
|
mapping = {".py": "Python", ".js": "JavaScript", ".jsx": "JavaScript/React", ".ts": "TypeScript", ".tsx": "TypeScript/React", ".go": "Go", ".rs": "Rust", ".java": "Java", ".rb": "Ruby", ".php": "PHP", ".cs": "C#", ".cpp": "C++", ".c": "C", ".swift": "Swift", ".kt": "Kotlin", ".tf": "Terraform", ".sh": "Shell"}
|
|
counts = Counter(mapping.get(Path(p).suffix.lower()) for p in file_paths)
|
|
counts.pop(None, None)
|
|
return dict(counts.most_common())
|
|
|
|
|
|
def _infer_frameworks(file_index: dict[str, RepoFile], file_paths: set[str]) -> list[str]:
|
|
frameworks = set()
|
|
pkg = file_index.get("package.json")
|
|
if pkg:
|
|
text = pkg.text.lower()
|
|
for name in ["react", "next", "vite", "express", "nestjs", "vue", "svelte"]:
|
|
if f'"{name}"' in text:
|
|
frameworks.add(name)
|
|
pyproject = file_index.get("pyproject.toml") or file_index.get("requirements.txt")
|
|
if pyproject:
|
|
text = pyproject.text.lower()
|
|
for name in ["django", "flask", "fastapi", "pydantic", "pytest"]:
|
|
if name in text:
|
|
frameworks.add(name)
|
|
if "go.mod" in file_paths:
|
|
frameworks.add("Go modules")
|
|
if "Cargo.toml" in file_paths:
|
|
frameworks.add("Cargo")
|
|
return sorted(frameworks)
|
|
|
|
|
|
def _infer_entrypoints(file_paths: set[str], file_index: dict[str, RepoFile]) -> list[str]:
|
|
candidates = []
|
|
names = {"main.py", "app.py", "server.py", "index.js", "server.js", "main.go", "cmd/main.go", "Dockerfile", "package.json", "pyproject.toml"}
|
|
for p in sorted(file_paths):
|
|
if Path(p).name in names or p in names or p.startswith("cmd/"):
|
|
candidates.append(p)
|
|
return candidates[:30]
|
|
|
|
|
|
def _major_components(file_paths: set[str]) -> list[dict[str, Any]]:
|
|
top = Counter(p.split("/", 1)[0] for p in file_paths if "/" in p)
|
|
return [{"path": name, "files": count, "category": classify_path(name + "/")} for name, count in top.most_common(20)]
|
|
|
|
|
|
def _infer_external_services(files: list[RepoFile]) -> list[str]:
|
|
services = set()
|
|
patterns = {"postgres": "PostgreSQL", "mysql": "MySQL", "redis": "Redis", "mongodb": "MongoDB", "s3": "AWS S3", "stripe": "Stripe", "sendgrid": "SendGrid", "slack": "Slack", "openai": "OpenAI", "anthropic": "Anthropic", "kafka": "Kafka", "rabbitmq": "RabbitMQ"}
|
|
for f in files:
|
|
text = f.text.lower()
|
|
for needle, label in patterns.items():
|
|
if needle in text:
|
|
services.add(label)
|
|
return sorted(services)
|
|
|
|
|
|
def _infer_data_flow(files: list[RepoFile], external_services: list[str]) -> str:
|
|
source_files = [f for f in files if f.category == "source"]
|
|
api_terms = sum(1 for f in source_files for term in ("router", "route", "controller", "handler", "endpoint") if term in f.text.lower())
|
|
db_terms = sum(1 for f in source_files for term in ("database", "db", "query", "model", "schema", "migration") if term in f.text.lower())
|
|
parts = []
|
|
if api_terms:
|
|
parts.append("request/handler style application code")
|
|
if db_terms:
|
|
parts.append("database/model access paths")
|
|
if external_services:
|
|
parts.append("external services: " + ", ".join(external_services[:8]))
|
|
return "; ".join(parts) if parts else "Data flow could not be confidently inferred from deterministic heuristics."
|
|
|
|
|
|
def _infer_build_test_deploy(file_paths: set[str], file_index: dict[str, RepoFile]) -> dict[str, list[str]]:
|
|
build = [p for p in file_paths if Path(p).name.lower() in {"package.json", "pyproject.toml", "makefile", "pom.xml", "build.gradle", "go.mod", "cargo.toml"}]
|
|
test = [p for p in file_paths if classify_path(p) == "tests"][:30]
|
|
deploy = [p for p in file_paths if classify_path(p) in {"CI/CD", "Docker/container", "infra/IaC"}][:50]
|
|
return {"build": sorted(build), "test": sorted(test), "deploy": sorted(deploy)}
|
|
|
|
|
|
def _redact_secrets(text: str) -> str:
|
|
redacted = text
|
|
for label, pattern in SECRET_REGEXES:
|
|
def repl(match: re.Match[str]) -> str:
|
|
if label == "Generic assigned secret" and match.lastindex and match.lastindex >= 2:
|
|
return f"{match.group(1)}=<REDACTED>"
|
|
value = match.group(0)
|
|
return value[:4] + "<REDACTED>" + value[-4:] if len(value) > 12 else "<REDACTED>"
|
|
redacted = pattern.sub(repl, redacted)
|
|
return redacted
|
|
|
|
|
|
def _worst_severity(severities: list[str]) -> str:
|
|
order = {"critical": 4, "high": 3, "medium": 2, "low": 1}
|
|
return max(severities or ["low"], key=lambda s: order.get(s, 0))
|
|
|
|
|
|
def _slug(text: str) -> str:
|
|
return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")[:90]
|
|
|
|
|
|
def _render_markdown(analysis: dict[str, Any]) -> str:
|
|
repo = analysis["repository"]
|
|
score = analysis["scorecard"]
|
|
lines = [
|
|
f"# Repository inspection: {repo.get('full_name')}",
|
|
"",
|
|
f"Selected ref: `{repo.get('selected_ref')}` ",
|
|
f"Production-readiness score: **{score['total']}/100**",
|
|
"",
|
|
"## Scorecard",
|
|
"",
|
|
]
|
|
for name, value in score["sub_scores"].items():
|
|
lines.append(f"- **{name}**: {value}/100")
|
|
lines += ["", "## Architecture summary", ""]
|
|
arch = analysis["architecture_summary"]
|
|
lines.append(f"- Languages: {', '.join(f'{k} ({v})' for k, v in arch['languages'].items()) or 'not confidently inferred'}")
|
|
lines.append(f"- Frameworks/tools: {', '.join(arch['frameworks']) or 'not confidently inferred'}")
|
|
lines.append(f"- Entrypoints: {', '.join(f'`{p}`' for p in arch['entrypoints'][:12]) or 'not detected'}")
|
|
lines.append(f"- Data flow: {arch['data_flow']}")
|
|
lines.append(f"- External services: {', '.join(arch['external_services']) or 'none detected by heuristics'}")
|
|
lines += ["", "## Repository map", ""]
|
|
for cat, count in analysis["repo_map"]["counts_by_category"].items():
|
|
lines.append(f"- {cat}: {count}")
|
|
lines += ["", "## Findings", ""]
|
|
if not analysis["findings"]:
|
|
lines.append("No evidence-backed risks were detected by the configured heuristics.")
|
|
for f in analysis["findings"][:120]:
|
|
lines.append(f"### [{f['severity']}] {f['title']}")
|
|
lines.append(f"Category: {f['category']} ")
|
|
lines.append(f"{f['description']}")
|
|
for c in f.get("citations", []):
|
|
lines.append(_format_citation_line(c))
|
|
lines.append("")
|
|
lines += ["", "## Recommended fixes", ""]
|
|
for r in analysis["recommendations"][:80]:
|
|
lines.append(f"### [{r['severity']}] {r['title']}")
|
|
lines.append(f"Impact: {r['impact']} · Effort: {r['effort']} ")
|
|
lines.append(r["description"])
|
|
for c in r.get("citations", [])[:5]:
|
|
lines.append(_format_citation_line(c))
|
|
lines.append("")
|
|
if analysis["skipped_files"]:
|
|
lines += ["", "## Skipped/limited files", ""]
|
|
for item in analysis["skipped_files"][:200]:
|
|
lines.append(f"- `{item.get('path')}` — {item.get('reason')}: {item.get('detail')}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _runtime_skills_root(ctx: RunContext[Any]) -> str:
|
|
workspace = getattr(ctx, "_workspace", None)
|
|
prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
|
|
if not prefixes:
|
|
outputs_prefix = getattr(workspace, "outputs_prefix", None)
|
|
prefixes = (outputs_prefix or "outputs/",)
|
|
prefix = str(prefixes[0]).strip("/")
|
|
return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
|
|
|
|
|
|
def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
|
|
root = Path(__file__).parent / "skills"
|
|
if not root.exists():
|
|
return []
|
|
runtime_skills_root = _runtime_skills_root(ctx)
|
|
uploads: list[tuple[str, bytes]] = []
|
|
for path in root.rglob("*"):
|
|
if path.is_file():
|
|
rel = path.relative_to(root).as_posix()
|
|
uploads.append((runtime_skills_root + rel, path.read_bytes()))
|
|
if uploads:
|
|
backend.upload_files(uploads)
|
|
return [runtime_skills_root]
|
|
return []
|