"""SSH key service utilities for preparing keys for container use.""" import logging import os from pathlib import Path from cryptography.fernet import Fernet from src.config import Settings logger = logging.getLogger(__name__) def _get_fernet() -> Fernet: """Generate a valid Fernet key from the session secret.""" import base64 import hashlib settings = Settings() key_bytes = hashlib.sha256(settings.session_secret.encode()).digest() key = base64.urlsafe_b64encode(key_bytes) return Fernet(key) def prepare_ssh_key_files( instance_dir: str, ssh_key, subdir: str = ".ssh", uid: int | None = None, gid: int | None = None, ) -> str: """Decrypt and write SSH key files to instance directory for container mounting. Args: instance_dir: Path to instance directory ssh_key: SSHKey model instance with encrypted private key subdir: Subdirectory within instance_dir to write to (default: ".ssh") uid: Optional UID to own the files (for bind-mount into non-root container) gid: Optional GID to own the files Returns: Path to the .ssh directory """ ssh_dir = Path(instance_dir) / subdir ssh_dir.mkdir(parents=True, exist_ok=True) # Decrypt private key fernet = _get_fernet() private_key = fernet.decrypt(ssh_key.private_key_encrypted.encode()).decode() # Write private key with restricted permissions private_key_path = ssh_dir / "id_ed25519" private_key_path.write_text(private_key) os.chmod(private_key_path, 0o600) # Write public key public_key_path = ssh_dir / "id_ed25519.pub" public_key_path.write_text(ssh_key.public_key) os.chmod(public_key_path, 0o644) # Write SSH config config_path = ssh_dir / "config" config_content = """Host * StrictHostKeyChecking no UserKnownHostsFile /dev/null IdentityFile ~/.ssh/id_ed25519 IdentitiesOnly yes """ config_path.write_text(config_content) os.chmod(config_path, 0o644) # Set ownership to target container user if requested if uid is not None or gid is not None: effective_uid = uid if uid is not None else -1 effective_gid = gid if gid is not None else -1 try: os.chown(ssh_dir, effective_uid, effective_gid) os.chown(private_key_path, effective_uid, effective_gid) os.chown(public_key_path, effective_uid, effective_gid) os.chown(config_path, effective_uid, effective_gid) logger.debug( "Set SSH key ownership to uid=%s gid=%s for %s", effective_uid, effective_gid, ssh_dir, ) except PermissionError as exc: logger.warning( "Cannot chown SSH keys to uid=%s gid=%s (running as uid=%s): %s", effective_uid, effective_gid, os.getuid(), exc, ) return str(ssh_dir) def cleanup_ssh_key_files(instance_dir: str) -> None: """Remove temporary SSH key files from instance directory. Args: instance_dir: Path to instance directory """ ssh_dir = Path(instance_dir) / ".ssh" if ssh_dir.exists(): for file_path in ssh_dir.iterdir(): file_path.unlink() ssh_dir.rmdir()