Orchestrator/shared/AI-Orchestrator/scripts/pat_manager/secret_updater.py

"""GitHub Actions secret updater using the GitHub API.

Updates repository secrets by encrypting values with the repository's
public key using libsodium sealed box encryption (NaCl).
"""

from __future__ import annotations

import base64
import logging

import httpx
from nacl.public import PublicKey, SealedBox

from .errors import SecretUpdateError

logger = logging.getLogger(__name__)

GITHUB_API_BASE = "https://api.github.com"


class SecretUpdater:
    """Updates GitHub Actions repository secrets via the GitHub API.

    Uses libsodium sealed box encryption to encrypt secret values before
    sending them to the GitHub API.

    Attributes:
        github_token: GitHub PAT with repo scope for secret management.
        repo_owner: Repository owner (user or organization).
        repo_name: Repository name.
    """

    def __init__(self, github_token: str, repo_owner: str, repo_name: str) -> None:
        self._github_token = github_token
        self._repo_owner = repo_owner
        self._repo_name = repo_name

    def _headers(self) -> dict[str, str]:
        """Build authorization headers for GitHub API requests."""
        return {
            "Authorization": f"Bearer {self._github_token}",
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        }

    def _encrypt_secret(self, public_key_b64: str, secret_value: str) -> str:
        """Encrypt a secret value using the repository's public key.

        Args:
            public_key_b64: Base64-encoded repository public key from GitHub API.
            secret_value: The plaintext secret value to encrypt.

        Returns:
            Base64-encoded encrypted value ready for the GitHub API.
        """
        public_key_bytes = base64.b64decode(public_key_b64)
        public_key = PublicKey(public_key_bytes)
        sealed_box = SealedBox(public_key)
        encrypted = sealed_box.encrypt(secret_value.encode("utf-8"))
        return base64.b64encode(encrypted).decode("utf-8")

    async def update_secret(self, secret_name: str, secret_value: str) -> bool:
        """Update a GitHub Actions repository secret.

        Fetches the repository's public key, encrypts the secret value using
        libsodium sealed box encryption, and updates the secret via the GitHub API.

        On failure, the previous secret value remains unchanged (GitHub API is atomic).

        Args:
            secret_name: Name of the secret to update.
            secret_value: New plaintext value for the secret.

        Returns:
            True if the secret was updated successfully.

        Raises:
            SecretUpdateError: If the update fails for any reason.
                The error message never contains the secret value.
        """
        base_url = (
            f"{GITHUB_API_BASE}/repos/{self._repo_owner}/{self._repo_name}"
        )

        async with httpx.AsyncClient() as client:
            # Step 1: Get the repository's public key for secret encryption
            try:
                key_response = await client.get(
                    f"{base_url}/actions/secrets/public-key",
                    headers=self._headers(),
                )
            except httpx.HTTPError as exc:
                raise SecretUpdateError(
                    secret_name,
                    f"Failed to fetch repository public key: {type(exc).__name__}",
                ) from exc

            if key_response.status_code != 200:
                raise SecretUpdateError(
                    secret_name,
                    f"Failed to fetch repository public key "
                    f"(HTTP {key_response.status_code})",
                )

            key_data = key_response.json()
            public_key_b64 = key_data["key"]
            key_id = key_data["key_id"]

            # Step 2: Encrypt the secret value using the public key
            try:
                encrypted_value = self._encrypt_secret(public_key_b64, secret_value)
            except Exception as exc:
                raise SecretUpdateError(
                    secret_name,
                    f"Failed to encrypt secret value: {type(exc).__name__}",
                ) from exc

            # Step 3: Update the secret via PUT request
            try:
                put_response = await client.put(
                    f"{base_url}/actions/secrets/{secret_name}",
                    headers=self._headers(),
                    json={
                        "encrypted_value": encrypted_value,
                        "key_id": key_id,
                    },
                )
            except httpx.HTTPError as exc:
                raise SecretUpdateError(
                    secret_name,
                    f"Failed to update secret: {type(exc).__name__}",
                ) from exc

            # GitHub returns 201 (created) or 204 (updated) on success
            if put_response.status_code not in (201, 204):
                raise SecretUpdateError(
                    secret_name,
                    f"Failed to update secret (HTTP {put_response.status_code})",
                )

            logger.info("Successfully updated GitHub secret '%s'", secret_name)
            return True