86 lines
2.8 KiB
Python
86 lines
2.8 KiB
Python
"""
|
|
Shared resource helpers for the coding A2A agent.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from urllib.parse import quote
|
|
|
|
from .config import GitProvider
|
|
|
|
|
|
def ensure_model_prefix(model_name: str) -> str:
|
|
return model_name if ":" in model_name else f"openai:{model_name}"
|
|
|
|
|
|
def detect_git_provider(repo_url: str) -> GitProvider:
|
|
lowered = repo_url.lower()
|
|
if "github" in lowered:
|
|
return GitProvider.github
|
|
if "gitlab" in lowered:
|
|
return GitProvider.gitlab
|
|
if "gitea" in lowered or ":3000/" in lowered:
|
|
return GitProvider.gitea
|
|
return GitProvider.generic
|
|
|
|
|
|
def build_authenticated_repo_url(
|
|
repo_url: str,
|
|
username: Optional[str] = None,
|
|
password: Optional[str] = None,
|
|
token: Optional[str] = None,
|
|
) -> str:
|
|
if not repo_url.startswith(("http://", "https://")):
|
|
return repo_url
|
|
provider = detect_git_provider(repo_url)
|
|
if token and not username:
|
|
if provider == GitProvider.github:
|
|
username = "x-access-token"
|
|
elif provider == GitProvider.gitlab:
|
|
username = "oauth2"
|
|
else:
|
|
username = "git"
|
|
password = token
|
|
elif token and username and not password:
|
|
password = token
|
|
if not username or not password:
|
|
return repo_url
|
|
encoded_user = quote(username, safe="")
|
|
encoded_password = quote(password, safe="")
|
|
if repo_url.startswith("https://"):
|
|
return repo_url.replace("https://", f"https://{encoded_user}:{encoded_password}@", 1)
|
|
return repo_url.replace("http://", f"http://{encoded_user}:{encoded_password}@", 1)
|
|
|
|
|
|
def safe_workspace_path(root_dir: str, relative_path: str, allowed_paths: Optional[list[str]] = None) -> Path:
|
|
root = Path(root_dir).resolve()
|
|
candidate = (root / relative_path).resolve()
|
|
if candidate != root and root not in candidate.parents:
|
|
raise ValueError(f"path escapes workspace: {relative_path}")
|
|
if ".git" in candidate.parts:
|
|
raise ValueError("access to .git is not allowed")
|
|
if ".github" in candidate.parts and "workflows" in candidate.parts:
|
|
raise ValueError("access to .github/workflows is not allowed")
|
|
if allowed_paths:
|
|
normalized = candidate.relative_to(root).as_posix()
|
|
if not any(
|
|
normalized == path.strip("/")
|
|
or normalized.startswith(f"{path.strip('/')}/")
|
|
for path in allowed_paths
|
|
):
|
|
raise ValueError(f"path outside allowed_paths: {relative_path}")
|
|
return candidate
|
|
|
|
|
|
def summarize_resources(resources: dict) -> dict:
|
|
summary = {}
|
|
for key, value in resources.items():
|
|
if not value:
|
|
continue
|
|
if isinstance(value, dict):
|
|
summary[key] = sorted(value.keys())
|
|
else:
|
|
summary[key] = str(type(value).__name__)
|
|
return summary
|