import os import pygit2 from pygit2 import RemoteCallbacks, CredentialType, Username, UserPass, Keypair from pygit2.callbacks import _Credentials from pygit2.enums import BranchType, FetchPrune, CheckoutStrategy as CS from pillar_tool.util import Config class RepoCallbacks(RemoteCallbacks): def __init__(self, config: Config): super().__init__() self.config = config def credentials( self, url: str, username_from_url: str | None, allowed_types: CredentialType, ) -> Username | UserPass | Keypair: # compute allowed methods allowed = [ 2**i for i,v in enumerate(reversed(bin(15)[2:])) if int(v) ] if CredentialType.SSH_KEY.value in allowed: print("cred ssh_key") return Keypair( username=self.config.git.state_repo_user, privkey=self.config.git.state_repo_keyfile, pubkey=self.config.git.state_repo_pubkeyfile, passphrase=None ) elif CredentialType.USERNAME.value in allowed: print("cred username") return Username(self.config.git.state_repo_user) print(f"The remote requested invalid credentials: {allowed}") raise RuntimeError(f"The remote requested invalid credentials: {allowed}") def checkout_remote_branch(config: Config, branch_name: str) -> None: """Checkout a remote branch from the state repository. Verifies that the state repository exists and has an 'origin' remote, then checks out the specified remote branch. Args: config: Configuration object containing the git state repository path. branch_name: Name of the remote branch to checkout. Raises: FileNotFoundError: If the state repository directory does not exist. ValueError: If the repository cannot be opened, 'origin' remote is missing, or the specified branch does not exist. """ # create an instance of the RepositoryCallback class cbs = RepoCallbacks(config) # check if the repository actually exists if not os.path.isdir(config.git.state_repo_path): # if the directory does not yet exist, clone the repository try: print("cloning state repo") os.makedirs(os.path.dirname(config.git.state_repo_path), mode=0o700, exist_ok=True) repository = pygit2.clone_repository(config.git.state_repo_remote, config.git.state_repo_path, callbacks=cbs, depth=1) except Exception as e: print(f"Failed to clone state repo: {e}") raise ValueError(f"Unable to clone the states repository: {e}") else: # directory exists, so attempt to open the repository try: repository = pygit2.Repository(config.git.state_repo_path) except Exception: raise ValueError(f"State repo at {config.git.state_repo_path} cannot be opened") # check whether this repository has a remote named origin # this only needs to happen when the repo has not just been cloned if "origin" not in repository.remotes: raise ValueError(f"No remote named origin in repo at {config.git.state_repo_path}") else: repository.remotes["origin"].fetch(prune=FetchPrune.PRUNE, depth=1, callbacks=cbs) # check if the requested branch exists try: branch_ref = repository.lookup_branch(f'origin/{branch_name}', BranchType.REMOTE) except KeyError: raise ValueError(f"Branch '{branch_name}' does not exist in the repository.") try: # checkout the remote branch with force # this should be done like this, since there should never be any change made in this clone of the repository repository.checkout(branch_ref, callbacks=cbs, strategy=CS.FORCE | CS.RECREATE_MISSING | CS.REMOVE_UNTRACKED) except Exception as exc: raise ValueError(f"Failed to checkout branch: {exc}")