"""Business logic for Azure DevOps connections and project mirrors.""" from __future__ import annotations from collections.abc import Awaitable, Callable from datetime import datetime, timedelta, timezone from typing import Optional from uuid import UUID import structlog from jose import JWTError, jwt from src.config import get_settings from src.core.exceptions import ForbiddenError, NotFoundError, ValidationError from src.core.security.encryption import decrypt_value, encrypt_value from src.domain.azure_user_mirror.models import WorkspaceUserAzureConnection from src.domain.azure_user_mirror.repository import ( ProjectUserAzureMirrorRepository, WorkspaceUserAzureConnectionRepository, ) from src.domain.projects.repository import ProjectRepository from src.integrations.azure_devops_client import AzureDevOpsClient, split_azure_repo_full_name logger = structlog.get_logger() _STATE_ALG = "HS256" _STATE_TTL_MIN = 30 def encode_azure_oauth_state( workspace_id: UUID, user_id: UUID, return_url: str, azure_org_url: str, import_project_id: Optional[UUID] = None, ) -> str: settings = get_settings() now = datetime.now(timezone.utc) payload = { "ws": str(workspace_id), "sub": str(user_id), "ru": return_url, "ou": azure_org_url.rstrip("/"), "iat": now, "exp": now + timedelta(minutes=_STATE_TTL_MIN), } if import_project_id: payload["ip"] = str(import_project_id) return jwt.encode(payload, settings.jwt_secret, algorithm=_STATE_ALG) def decode_azure_oauth_state(token: str) -> dict: settings = get_settings() try: payload = jwt.decode(token, settings.jwt_secret, algorithms=[_STATE_ALG]) except JWTError as e: raise ValidationError("Invalid or expired state token") from e ws = payload.get("ws") sub = payload.get("sub") org_url = payload.get("ou") if not ws or not sub or not isinstance(org_url, str) or not org_url.strip(): raise ValidationError("Invalid state payload") ip = payload.get("ip") return { "workspace_id": UUID(ws), "user_id": UUID(sub), "return_url": payload.get("ru") if isinstance(payload.get("ru"), str) else None, "azure_org_url": org_url.rstrip("/"), "import_project_id": UUID(ip) if isinstance(ip, str) and ip.strip() else None, } async def resolve_azure_connection_client( connection: WorkspaceUserAzureConnection, connection_repo: Optional[WorkspaceUserAzureConnectionRepository] = None, *, redirect_uri: Optional[str] = None, ) -> AzureDevOpsClient: """Return a client for PAT or OAuth connections, refreshing OAuth when possible.""" if (getattr(connection, "auth_type", "pat") or "pat") == "pat": return AzureDevOpsClient( org_url=connection.azure_org_url, pat=decrypt_value(connection.pat_ciphertext), ) if not connection.access_token_ciphertext: raise ValidationError("Azure DevOps OAuth connection is missing an access token.") access_token = decrypt_value(connection.access_token_ciphertext) expires_at = connection.token_expires_at if expires_at and expires_at.tzinfo is None: expires_at = expires_at.replace(tzinfo=timezone.utc) should_refresh = bool(expires_at and expires_at <= datetime.now(timezone.utc) + timedelta(minutes=5)) if should_refresh and connection.refresh_token_ciphertext and connection_repo is not None: token = await AzureDevOpsClient(org_url=connection.azure_org_url).refresh_token( refresh_token=decrypt_value(connection.refresh_token_ciphertext), redirect_uri=redirect_uri, ) next_access_token = token.get("access_token") or "" if next_access_token: next_refresh_token = token.get("refresh_token") or None expires_epoch = AzureDevOpsClient.token_expiry_epoch(token) await connection_repo.update_oauth_tokens( connection.id, access_token_ciphertext=encrypt_value(next_access_token), refresh_token_ciphertext=( encrypt_value(next_refresh_token) if next_refresh_token else None ), token_expires_at=( datetime.fromtimestamp(expires_epoch, timezone.utc) if expires_epoch is not None else None ), ) access_token = next_access_token return AzureDevOpsClient(org_url=connection.azure_org_url, access_token=access_token) class AzureUserMirrorService: def __init__( self, connection_repo: WorkspaceUserAzureConnectionRepository, mirror_repo: ProjectUserAzureMirrorRepository, project_repo: ProjectRepository, ): self._conn = connection_repo self._mirror = mirror_repo self._projects = project_repo def build_connect_url( self, *, workspace_id: UUID, user_id: UUID, azure_org_url: str, callback_return_url: str, redirect_uri: str, import_project_id: Optional[UUID] = None, ) -> tuple[str, str]: settings = get_settings() if ( not settings.azure_devops_oauth_client_id or not settings.azure_devops_oauth_client_secret ): raise ValidationError("Azure DevOps OAuth is not configured.") org_url = azure_org_url.strip().rstrip("/") if not org_url: raise ValidationError("azure_org_url is required.") state = encode_azure_oauth_state( workspace_id, user_id, callback_return_url, org_url, import_project_id=import_project_id, ) return AzureDevOpsClient(org_url=org_url).authorization_url( redirect_uri=redirect_uri, state=state, ), state async def exchange_callback( self, *, code: str, state: str, redirect_uri: str, ): state_payload = decode_azure_oauth_state(state) org_url = state_payload["azure_org_url"] client = AzureDevOpsClient(org_url=org_url) token = await client.exchange_code(code=code, redirect_uri=redirect_uri) access_token = token.get("access_token") or "" if not access_token: raise ValidationError("Azure DevOps did not return an access token.") refresh_token = token.get("refresh_token") or None authed = AzureDevOpsClient(org_url=org_url, access_token=access_token) profile = await authed.get_profile() username = str( profile.get("emailAddress") or profile.get("coreAttributes", {}).get("Email", {}).get("value") or profile.get("id") or "azure-devops-user" ) display_name = profile.get("displayName") or profile.get("publicAlias") expires_at = client.token_expiry_epoch(token) row = await self._conn.upsert_oauth( workspace_id=state_payload["workspace_id"], azure_org_url=org_url, username=username, display_name=display_name, access_token_ciphertext=encrypt_value(access_token), refresh_token_ciphertext=encrypt_value(refresh_token) if refresh_token else None, token_expires_at=( datetime.fromtimestamp(expires_at, timezone.utc) if expires_at is not None else None ), created_by=state_payload["user_id"], empty_pat_ciphertext=encrypt_value(""), ) return row, state_payload async def create_connection( self, *, workspace_id: UUID, user_id: UUID, azure_org_url: str, username: str, display_name: Optional[str], pat: str, ): org_url = azure_org_url.strip().rstrip("/") if not org_url: raise ValidationError("azure_org_url is required.") if not pat.strip(): raise ValidationError("PAT is required.") client = AzureDevOpsClient(org_url=org_url, pat=pat) try: await client.verify_pat() except Exception as exc: raise ValidationError( f"Could not authenticate with Azure DevOps using the provided PAT: {exc}" ) from exc return await self._conn.upsert( workspace_id=workspace_id, azure_org_url=org_url, username=username.strip(), display_name=(display_name or "").strip() or None, pat_ciphertext=encrypt_value(pat), created_by=user_id, ) async def list_connection_repositories(self, workspace_id: UUID, row_id: UUID) -> list[dict]: row = await self._conn.get_by_id(row_id) if not row or row.workspace_id != workspace_id: raise NotFoundError("Azure DevOps connection not found") if row.revoked_at is not None: raise ValidationError("Azure DevOps connection is revoked; reconnect with a new PAT.") client = await resolve_azure_connection_client(row, self._conn) repos = await client.list_repositories() out = [] for r in repos: project_name = (r.get("project") or {}).get("name") or "" repo_name = r.get("name") or "" if not project_name or not repo_name: continue full_name = f"{project_name}/{repo_name}" out.append({ "id": str(r.get("id") or ""), "full_name": full_name, "name": repo_name, "project": project_name, "default_branch": client.default_branch_name(r), "web_url": (r.get("remoteUrl") or "").replace( "/_git/", "/_git/" ), }) return out async def validate_source( self, *, workspace_id: UUID, connection_row_id: UUID, repo_full_name: str, branch: Optional[str], ) -> dict: row = await self._conn.get_by_id(connection_row_id) if not row or row.workspace_id != workspace_id or row.revoked_at is not None: raise ValidationError("Invalid or revoked Azure DevOps connection for this workspace.") project, repo = split_azure_repo_full_name(repo_full_name) client = await resolve_azure_connection_client(row, self._conn) try: repo_data = await client.get_repository(project, repo) except Exception as exc: raise ValidationError( f"Could not access repository '{repo_full_name}' in Azure DevOps: {exc}" ) from exc default_branch = client.default_branch_name(repo_data) return { "connection_row_id": row.id, "repo_full_name": repo_full_name.strip(), "branch": (branch or "").strip() or default_branch, "default_branch": default_branch, "repo_id": str(repo_data.get("id") or ""), } async def upsert_project_mirror( self, *, project_id: UUID, workspace_id: UUID, user_id: UUID, connection_row_id: UUID, target_repo_full_name: str, target_default_branch: str, enabled: bool, check_permission: Callable[[UUID, UUID, str], Awaitable[bool]], ): if not await check_permission(workspace_id, user_id, "repos.manage"): raise ForbiddenError() proj = await self._projects.get_by_id(project_id) if not proj or proj.workspace_id != workspace_id: raise NotFoundError("Project not found") conn = await self._conn.get_by_id(connection_row_id) if not conn or conn.workspace_id != workspace_id or conn.revoked_at is not None: raise ValidationError("Invalid or revoked Azure DevOps connection for this workspace.") source = await self.validate_source( workspace_id=workspace_id, connection_row_id=connection_row_id, repo_full_name=target_repo_full_name, branch=target_default_branch, ) mirror = await self._mirror.upsert( project_id=project_id, workspace_user_azure_connection_id=connection_row_id, target_repo_full_name=target_repo_full_name.strip(), target_default_branch=target_default_branch.strip() or "main", enabled=enabled, ) await self._sync_webhook(conn, mirror, repo_id=source.get("repo_id")) return mirror async def _sync_webhook(self, conn, mirror, *, repo_id: Optional[str] = None) -> None: settings = get_settings() secret = (settings.azure_devops_webhook_secret or "").strip() api_url = (settings.api_url or "").rstrip("/") client = await resolve_azure_connection_client(conn, self._conn) webhook_url = f"{api_url}/api/webhooks/azure-devops" try: old_wh_id = mirror.target_repo_webhook_id effective_repo_id = repo_id or mirror.target_repo_id or "" if not effective_repo_id: return if old_wh_id: try: await client.delete_service_hook(old_wh_id) await self._mirror.update_webhook_id( mirror.project_id, repo_id=effective_repo_id, webhook_id=None, ) except Exception: pass if not mirror.enabled: return if not secret or not api_url: return # We need the project ID (GUID) to create the service hook project_name, _ = split_azure_repo_full_name(mirror.target_repo_full_name) repo_data = await client.get_repository(project_name, mirror.target_repo_full_name.split("/", 1)[1]) az_project_id = str((repo_data.get("project") or {}).get("id") or "") if not az_project_id: return new_wh_id = await client.create_service_hook( webhook_url=webhook_url, token=secret, project_id=az_project_id, repo_id=effective_repo_id, ) await self._mirror.update_webhook_id( mirror.project_id, repo_id=effective_repo_id, webhook_id=new_wh_id or None, ) except Exception as exc: logger.warning( "azure_webhook_registration_failed", project_id=str(mirror.project_id), error=str(exc), ) async def delete_project_mirror( self, *, project_id: UUID, workspace_id: UUID, user_id: UUID, check_permission: Callable[[UUID, UUID, str], Awaitable[bool]], ) -> bool: if not await check_permission(workspace_id, user_id, "repos.manage"): raise ForbiddenError() mirror = await self._mirror.get_by_project_id(project_id) if not mirror: return False conn = await self._conn.get_by_id(mirror.workspace_user_azure_connection_id) if conn and conn.revoked_at is None and mirror.target_repo_webhook_id: try: client = await resolve_azure_connection_client(conn, self._conn) await client.delete_service_hook(mirror.target_repo_webhook_id) except Exception as exc: logger.warning("azure_webhook_delete_failed", error=str(exc)) await self._mirror.delete_by_project_id(project_id) return True async def cleanup_webhooks_for_connection(self, connection_id: UUID) -> None: """Deregister all service hook subscriptions before a connection is deleted.""" conn = await self._conn.get_by_id(connection_id) mirrors = await self._mirror.list_by_connection_id(connection_id) if not mirrors or not conn or conn.revoked_at is not None: return try: client = await resolve_azure_connection_client(conn, self._conn) except Exception: return for mirror in mirrors: if mirror.target_repo_webhook_id: try: await client.delete_service_hook(mirror.target_repo_webhook_id) except Exception as exc: logger.warning("azure_webhook_cleanup_failed", error=str(exc))