diff --git a/apps/api/app/git/__init__.py b/apps/api/app/git/__init__.py index 8ef66ce..6bed921 100644 --- a/apps/api/app/git/__init__.py +++ b/apps/api/app/git/__init__.py @@ -2,6 +2,7 @@ from app.git.credentials import AccessTokenCredential, CredentialStorage, GitCredential from app.git.provider import GitProvider +from app.git.ssh_key import SshKeyLifecycle, SshKeyPair from app.git.types import ConnectionStatus, CredentialKind, ProviderKind, SshKeyStatus __all__ = [ @@ -12,5 +13,7 @@ __all__ = [ "GitCredential", "GitProvider", "ProviderKind", + "SshKeyLifecycle", + "SshKeyPair", "SshKeyStatus", ] diff --git a/apps/api/app/git/ssh_key.py b/apps/api/app/git/ssh_key.py new file mode 100644 index 0000000..e918700 --- /dev/null +++ b/apps/api/app/git/ssh_key.py @@ -0,0 +1,89 @@ +"""SSH key pair generation and lifecycle management. + +Security rules: +- Private key material must never appear in logs, exceptions, ``__repr__``, + or test output. +- The ``encrypted_private_key`` field uses ``repr=False``. +""" + +import base64 +import uuid +from datetime import datetime, timezone + +from pydantic import BaseModel, Field + +from app.git.types import SshKeyStatus + + +def encrypt_private_key(raw: bytes) -> str: + """Placeholder encryption helper. + + base64-encodes *raw* until FN-009 delivers the real encryption backend. + """ + return base64.b64encode(raw).decode("ascii") + + +class SshKeyPair(BaseModel): + """An Ed25519 SSH key pair belonging to a repository connection.""" + + id: uuid.UUID = Field(default_factory=uuid.uuid4) + connection_id: uuid.UUID + public_key: str + encrypted_private_key: str = Field(repr=False) + status: SshKeyStatus = SshKeyStatus.generated + created_at: datetime = Field( + default_factory=lambda: datetime.now(timezone.utc) + ) + updated_at: datetime = Field( + default_factory=lambda: datetime.now(timezone.utc) + ) + revoked_at: datetime | None = None + + +class SshKeyLifecycle: + """Generate and transition SSH key pairs.""" + + @staticmethod + def generate(connection_id: uuid.UUID) -> SshKeyPair: + """Generate a new Ed25519 key pair for *connection_id*.""" + from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + ) + from cryptography.hazmat.primitives.serialization import ( + Encoding, + NoEncryption, + PrivateFormat, + PublicFormat, + ) + + private_key = Ed25519PrivateKey.generate() + public_key = private_key.public_key() + + public_key_pem = public_key.public_bytes( + Encoding.OpenSSH, PublicFormat.OpenSSH + ).decode("utf-8") + + private_key_pem = private_key.private_bytes( + Encoding.PEM, PrivateFormat.PKCS8, NoEncryption() + ) + + encrypted = encrypt_private_key(private_key_pem) + + return SshKeyPair( + connection_id=connection_id, + public_key=public_key_pem, + encrypted_private_key=encrypted, + status=SshKeyStatus.generated, + ) + + @staticmethod + def transition(key: SshKeyPair, new_status: SshKeyStatus) -> SshKeyPair: + """Update *key* status and timestamps. + + Sets ``revoked_at`` when transitioning to :attr:`SshKeyStatus.revoked`. + """ + key.status = new_status + key.updated_at = datetime.now(timezone.utc) + if new_status == SshKeyStatus.revoked: + key.revoked_at = datetime.now(timezone.utc) + return key