"""Business logic for user Bitbucket OAuth 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 logger = structlog.get_logger() from src.core.exceptions import ForbiddenError, NotFoundError, ValidationError from src.core.security.encryption import decrypt_value, encrypt_value from src.domain.bitbucket_user_mirror.repository import ( ProjectUserBitbucketMirrorRepository, WorkspaceUserBitbucketConnectionRepository, ) from src.domain.projects.repository import ProjectRepository from src.integrations.bitbucket_user_client import BitbucketUserClient _STATE_ALG = "HS256" _STATE_TTL_MIN = 30 def encode_bitbucket_oauth_state( workspace_id: UUID, user_id: UUID, return_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, "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_bitbucket_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") if not ws or not sub: 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, "import_project_id": UUID(ip) if isinstance(ip, str) and ip.strip() else None, } class BitbucketUserMirrorService: def __init__( self, connection_repo: WorkspaceUserBitbucketConnectionRepository, mirror_repo: ProjectUserBitbucketMirrorRepository, 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, callback_return_url: str, redirect_uri: str, import_project_id: Optional[UUID] = None, ) -> tuple[str, str]: settings = get_settings() if ( not settings.bitbucket_user_oauth_client_id or not settings.bitbucket_user_oauth_client_secret ): raise ValidationError("Bitbucket OAuth is not configured.") state = encode_bitbucket_oauth_state( workspace_id, user_id, callback_return_url, import_project_id=import_project_id, ) return ( BitbucketUserClient().authorization_url(redirect_uri=redirect_uri, state=state), state, ) async def exchange_callback( self, *, code: str, state: str, redirect_uri: str, ): state_payload = decode_bitbucket_oauth_state(state) client = BitbucketUserClient() token = await client.exchange_code(code=code, redirect_uri=redirect_uri) access_token = token.get("access_token") or "" if not access_token: raise ValidationError("Bitbucket did not return an access token.") refresh_token = token.get("refresh_token") or None authed = BitbucketUserClient(access_token=access_token) user = await authed.get_current_user() account_id = str(user.get("account_id") or user.get("uuid") or "") if not account_id: raise ValidationError("Bitbucket did not return an account id.") expires_at = client.token_expiry_epoch(token) row = await self._conn.upsert( workspace_id=state_payload["workspace_id"], bitbucket_account_id=account_id, username=str(user.get("username") or user.get("nickname") or ""), display_name=user.get("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"], ) return row, state_payload 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("Bitbucket connection not found") if row.revoked_at is not None: raise ValidationError("Bitbucket connection is revoked; reconnect Bitbucket.") repos = await BitbucketUserClient( access_token=decrypt_value(row.access_token_ciphertext) ).list_repositories() return [ { "uuid": str(r.get("uuid") or ""), "full_name": r.get("full_name") or "", "name": r.get("name") or "", "private": bool(r.get("is_private", True)), "default_branch": (r.get("mainbranch") or {}).get("name"), "web_url": ((r.get("links") or {}).get("html") or {}).get("href"), } for r in repos if r.get("full_name") ] 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 Bitbucket connection for this workspace.") client = BitbucketUserClient(access_token=decrypt_value(row.access_token_ciphertext)) repo = await client.get_repository(repo_full_name) default_branch = str((repo.get("mainbranch") or {}).get("name") or "main") return { "connection_row_id": row.id, "repo_full_name": repo_full_name.strip(), "branch": (branch or "").strip() or default_branch, "default_branch": default_branch, } 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 Bitbucket connection for this workspace.") 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_bitbucket_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) return mirror async def _sync_webhook(self, conn, mirror) -> None: settings = get_settings() secret = (settings.bitbucket_user_webhook_secret or "").strip() api_url = (settings.api_url or "").rstrip("/") access_token = decrypt_value(conn.access_token_ciphertext) client = BitbucketUserClient(access_token=access_token) webhook_url = f"{api_url}/api/webhooks/bitbucket-user" try: old_uuid = mirror.target_repo_webhook_uuid if old_uuid: try: await client.delete_repo_webhook(mirror.target_repo_full_name, old_uuid) await self._mirror.update_webhook_id(mirror.project_id, None) except Exception: pass if not mirror.enabled: return if not secret or not api_url: return new_uuid = await client.register_repo_webhook( mirror.target_repo_full_name, webhook_url, secret ) await self._mirror.update_webhook_id(mirror.project_id, new_uuid or None) except Exception as exc: logger.warning( "bitbucket_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_bitbucket_connection_id) if conn and conn.revoked_at is None and mirror.target_repo_webhook_uuid: try: access_token = decrypt_value(conn.access_token_ciphertext) await BitbucketUserClient(access_token=access_token).delete_repo_webhook( mirror.target_repo_full_name, mirror.target_repo_webhook_uuid ) except Exception as exc: logger.warning("bitbucket_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 mirror webhooks 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 access_token = decrypt_value(conn.access_token_ciphertext) client = BitbucketUserClient(access_token=access_token) for mirror in mirrors: if mirror.target_repo_webhook_uuid: try: await client.delete_repo_webhook( mirror.target_repo_full_name, mirror.target_repo_webhook_uuid ) except Exception as exc: logger.warning("bitbucket_webhook_cleanup_failed", error=str(exc))