Developer
f6003b75ca
- Create services/docker/compose.py — compose file generation and commands - Create services/docker/container.py — container lifecycle and queries - Create services/docker/config_staging.py — config folder file writing - Create services/docker/tunnel.py — Cloudflare tunnel management - Create services/docker/__init__.py — barrel exports - Delete services/docker.py (replaced by package) - All imports in api/tool_instances.py remain functional Quality gates: Python syntax check (pass), imports verified Refs: repo-restructure Task 3.3
146 lines
4.2 KiB
Python
146 lines
4.2 KiB
Python
import uuid
|
|
from datetime import datetime
|
|
|
|
from cryptography.fernet import Fernet
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.auth.dependencies import get_current_user, get_db_session
|
|
from src.config import Settings
|
|
from src.models.ssh_key import SSHKey
|
|
from src.models.user import User
|
|
from src.schemas.ssh_key import SSHKeyCreate, SSHKeyResponse
|
|
|
|
router = APIRouter(prefix="/ssh-keys", tags=["ssh-keys"])
|
|
|
|
|
|
|
|
def _get_fernet() -> Fernet:
|
|
"""Generate a valid Fernet key from the session secret."""
|
|
import base64
|
|
import hashlib
|
|
|
|
settings = Settings()
|
|
# Derive a 32-byte key from the session secret using SHA256
|
|
key_bytes = hashlib.sha256(settings.session_secret.encode()).digest()
|
|
# Base64 encode it for Fernet (must be 32 url-safe base64-encoded bytes)
|
|
key = base64.urlsafe_b64encode(key_bytes)
|
|
return Fernet(key)
|
|
|
|
|
|
def generate_ssh_key_pair() -> tuple[str, str]:
|
|
"""Generate a new Ed25519 SSH key pair.
|
|
|
|
Returns:
|
|
Tuple of (private_key, public_key) as strings.
|
|
"""
|
|
private_key = Ed25519PrivateKey.generate()
|
|
public_key = private_key.public_key()
|
|
|
|
private_bytes = private_key.private_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PrivateFormat.OpenSSH,
|
|
encryption_algorithm=serialization.NoEncryption(),
|
|
)
|
|
|
|
public_bytes = public_key.public_bytes(
|
|
encoding=serialization.Encoding.OpenSSH,
|
|
format=serialization.PublicFormat.OpenSSH,
|
|
)
|
|
|
|
return private_bytes.decode("utf-8"), public_bytes.decode("utf-8")
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
response_model=SSHKeyResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
summary="Create SSH key",
|
|
description="Generate a new Ed25519 SSH key pair for the authenticated user.",
|
|
)
|
|
async def create_ssh_key(
|
|
data: SSHKeyCreate,
|
|
user: User = Depends(get_current_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> SSHKey:
|
|
"""Create a new SSH key pair.
|
|
|
|
Args:
|
|
data: SSH key creation data including the key name.
|
|
user_id: ID of the authenticated user.
|
|
session: Database session.
|
|
|
|
Returns:
|
|
The newly created SSH key with public key exposed.
|
|
"""
|
|
private_key, public_key = generate_ssh_key_pair()
|
|
|
|
fernet = _get_fernet()
|
|
encrypted_private = fernet.encrypt(private_key.encode()).decode()
|
|
|
|
ssh_key = SSHKey(
|
|
name=data.name,
|
|
public_key=public_key,
|
|
private_key_encrypted=encrypted_private,
|
|
user_id=user.id,
|
|
)
|
|
session.add(ssh_key)
|
|
await session.commit()
|
|
await session.refresh(ssh_key)
|
|
return ssh_key
|
|
|
|
|
|
@router.get(
|
|
"",
|
|
response_model=list[SSHKeyResponse],
|
|
summary="List SSH keys",
|
|
description="List all SSH keys for the authenticated user.",
|
|
)
|
|
async def list_ssh_keys(
|
|
user: User = Depends(get_current_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> list[SSHKey]:
|
|
"""List all SSH keys for the authenticated user.
|
|
|
|
Args:
|
|
user_id: ID of the authenticated user.
|
|
session: Database session.
|
|
|
|
Returns:
|
|
List of SSH keys owned by the user.
|
|
"""
|
|
result = await session.execute(select(SSHKey).where(SSHKey.user_id == user.id))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
@router.delete(
|
|
"/{key_id}",
|
|
status_code=status.HTTP_204_NO_CONTENT,
|
|
summary="Delete SSH key",
|
|
description="Delete an SSH key by ID.",
|
|
)
|
|
async def delete_ssh_key(
|
|
key_id: uuid.UUID,
|
|
user: User = Depends(get_current_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> None:
|
|
"""Delete an SSH key.
|
|
|
|
Args:
|
|
key_id: UUID of the SSH key to delete.
|
|
user_id: ID of the authenticated user.
|
|
session: Database session.
|
|
|
|
Returns:
|
|
None with 204 status code.
|
|
"""
|
|
ssh_key = await session.get(SSHKey, key_id)
|
|
if ssh_key is None or ssh_key.user_id != user.id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="ssh key not found")
|
|
|
|
await session.delete(ssh_key)
|
|
await session.commit()
|