"""Azure DevOps client for customer-owned repositories.""" from __future__ import annotations import time import urllib.parse from typing import Any, Optional import httpx from src.config import get_settings AZURE_API_VERSION = "7.1" AZURE_CLIENT_ASSERTION_TYPE = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" AZURE_JWT_BEARER_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:jwt-bearer" class AzureDevOpsClient: def __init__(self, *, org_url: str, pat: Optional[str] = None, access_token: Optional[str] = None): self._org_url = org_url.rstrip("/") self._pat = pat self._access_token = access_token def _request_kwargs(self) -> dict[str, Any]: if self._access_token: return {"headers": {"Authorization": f"Bearer {self._access_token}"}} if self._pat: return {"auth": ("", self._pat)} raise ValueError("Azure DevOps client requires a PAT or OAuth access token.") async def verify_pat(self) -> dict[str, Any]: """Verify the PAT works by fetching the authenticated user profile.""" url = f"https://app.vssps.visualstudio.com/_apis/profile/profiles/me?api-version={AZURE_API_VERSION}" async with httpx.AsyncClient(timeout=30.0) as client: r = await client.get(url, **self._request_kwargs()) r.raise_for_status() return r.json() async def get_profile(self) -> dict[str, Any]: """Fetch the authenticated user profile for PAT or OAuth credentials.""" url = f"https://app.vssps.visualstudio.com/_apis/profile/profiles/me?api-version={AZURE_API_VERSION}" async with httpx.AsyncClient(timeout=30.0) as client: r = await client.get(url, **self._request_kwargs()) r.raise_for_status() return r.json() async def list_repositories(self) -> list[dict[str, Any]]: """List all git repositories across all projects in the org.""" url = f"{self._org_url}/_apis/git/repositories?api-version={AZURE_API_VERSION}" out: list[dict[str, Any]] = [] async with httpx.AsyncClient(timeout=120.0) as client: while url: r = await client.get(url, **self._request_kwargs()) r.raise_for_status() data = r.json() out.extend(data.get("value") or []) # Azure DevOps REST API uses continuationToken for pagination continuation = r.headers.get("x-ms-continuationtoken") if continuation: sep = "&" if "?" in url else "?" url = f"{url}{sep}continuationToken={urllib.parse.quote(continuation)}" else: url = None return out async def get_repository(self, project: str, repo: str) -> dict[str, Any]: url = ( f"{self._org_url}/{urllib.parse.quote(project)}/_apis/git/repositories" f"/{urllib.parse.quote(repo)}?api-version={AZURE_API_VERSION}" ) async with httpx.AsyncClient(timeout=30.0) as client: r = await client.get(url, **self._request_kwargs()) r.raise_for_status() return r.json() def authenticated_clone_url(self, project: str, repo: str) -> str: # Strip scheme from org_url to rebuild with credentials parsed = urllib.parse.urlparse(self._org_url) host_path = parsed.netloc + parsed.path secret = self._pat or self._access_token or "" encoded_secret = urllib.parse.quote(secret, safe="") username = "pat" if self._pat else "oauth2" return ( f"https://{username}:{encoded_secret}@{host_path}/" f"{urllib.parse.quote(project)}/_git/{urllib.parse.quote(repo)}" ) def authorization_url(self, *, redirect_uri: str, state: str) -> str: settings = get_settings() query = { "client_id": settings.azure_devops_oauth_client_id, "response_type": "Assertion", "state": state, "scope": settings.azure_devops_oauth_scope, "redirect_uri": redirect_uri, } return f"{settings.azure_devops_oauth_authorize_url}?{urllib.parse.urlencode(query)}" async def exchange_code(self, *, code: str, redirect_uri: str) -> dict[str, Any]: settings = get_settings() data = { "client_assertion_type": AZURE_CLIENT_ASSERTION_TYPE, "client_assertion": settings.azure_devops_oauth_client_secret, "grant_type": AZURE_JWT_BEARER_GRANT_TYPE, "assertion": code, "redirect_uri": redirect_uri, } async with httpx.AsyncClient(timeout=30.0) as client: r = await client.post(settings.azure_devops_oauth_token_url, data=data) r.raise_for_status() return r.json() async def refresh_token( self, *, refresh_token: str, redirect_uri: Optional[str] = None, ) -> dict[str, Any]: settings = get_settings() data = { "client_assertion_type": AZURE_CLIENT_ASSERTION_TYPE, "client_assertion": settings.azure_devops_oauth_client_secret, "grant_type": "refresh_token", "assertion": refresh_token, } if redirect_uri: data["redirect_uri"] = redirect_uri async with httpx.AsyncClient(timeout=30.0) as client: r = await client.post(settings.azure_devops_oauth_token_url, data=data) r.raise_for_status() return r.json() @staticmethod def token_expiry_epoch(token: dict[str, Any]) -> Optional[int]: raw_expires_in = token.get("expires_in") try: return int(time.time()) + int(raw_expires_in) except (TypeError, ValueError): return None async def create_service_hook( self, *, webhook_url: str, token: str, project_id: str, repo_id: str, ) -> str: """Register a git.push Service Hook for a specific repository. Returns subscription ID.""" url = f"{self._org_url}/_apis/hooks/subscriptions?api-version={AZURE_API_VERSION}" payload = { "publisherId": "tfs", "eventType": "git.push", "resourceVersion": "1.0", "consumerId": "webHooks", "consumerActionId": "httpRequest", "publisherInputs": { "projectId": project_id, "repository": repo_id, "branch": "", "pushedBy": "", }, "consumerInputs": { "url": webhook_url, "httpHeaders": f"X-Azure-Token: {token}", "messagesToSend": "none", "detailedMessagesToSend": "all", }, } async with httpx.AsyncClient(timeout=30.0) as client: r = await client.post(url, json=payload, **self._request_kwargs()) r.raise_for_status() return str(r.json().get("id", "")) async def delete_service_hook(self, subscription_id: str) -> None: """Delete a Service Hook subscription. Silently ignores 404 (already gone).""" url = ( f"{self._org_url}/_apis/hooks/subscriptions/{subscription_id}" f"?api-version={AZURE_API_VERSION}" ) async with httpx.AsyncClient(timeout=30.0) as client: r = await client.delete(url, **self._request_kwargs()) if r.status_code == 404: return r.raise_for_status() @staticmethod def default_branch_name(repo: dict[str, Any]) -> str: raw = (repo.get("defaultBranch") or "refs/heads/main").strip() return raw.removeprefix("refs/heads/") or "main" def split_azure_repo_full_name(full_name: str) -> tuple[str, str]: """Split 'Project/RepoName' into (project, repo). Raises ValueError if malformed.""" raw = (full_name or "").strip().strip("/") parts = raw.split("/", 1) if len(parts) != 2 or not parts[0] or not parts[1]: raise ValueError("Azure DevOps repository must be in Project/RepoName form.") return parts[0], parts[1]