feat: implement SSH key management

- Add backend API endpoints for SSH key CRUD (POST, GET, DELETE)
- Implement Ed25519 key generation with Fernet-encrypted private keys
- Add frontend SSH keys page with generate, list, and delete functionality
- Include copy-to-clipboard for public keys
- Add responsive CSS styles for key cards
- Register ssh_keys router in main.py
- Add basic auth tests for SSH key endpoints

Quality gates: ruff ✓, mypy ✓, typecheck ✓, lint ✓
This commit is contained in:
Fusion
2026-05-18 14:44:21 +02:00
parent be81aa1c8b
commit a441ea2fac
24 changed files with 500 additions and 369 deletions
+130
View File
@@ -0,0 +1,130 @@
import uuid
from datetime import datetime
from typing import Annotated
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from fastapi import APIRouter, Cookie, Depends, HTTPException, status
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.auth.jwt_service import decode_access_token
from src.config import Settings
from src.database import SessionLocal
from src.models.ssh_key import SSHKey
from src.models.user import User
router = APIRouter(prefix="/ssh-keys", tags=["ssh-keys"])
async def get_db_session():
async with SessionLocal() as session:
yield session
async def get_current_user_id(
access_token: Annotated[str | None, Cookie()] = None,
) -> uuid.UUID:
if not access_token:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing access token")
try:
claims = decode_access_token(settings=Settings(), token=access_token)
return uuid.UUID(str(claims["sub"]))
except Exception:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid access token")
async def _get_user(session: AsyncSession, user_id: uuid.UUID) -> User:
user = await session.get(User, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="user not found")
return user
def _get_fernet() -> Fernet:
settings = Settings()
key = settings.jwt_secret[:32].ljust(32, "=")
return Fernet(key.encode())
def generate_ssh_key_pair() -> tuple[str, str]:
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")
class SSHKeyCreate(BaseModel):
name: str
class SSHKeyResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
name: str
public_key: str
created_at: datetime
@router.post("", response_model=SSHKeyResponse, status_code=status.HTTP_201_CREATED)
async def create_ssh_key(
data: SSHKeyCreate,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> SSHKey:
user = await _get_user(session, user_id)
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])
async def list_ssh_keys(
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> list[SSHKey]:
user = await _get_user(session, user_id)
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)
async def delete_ssh_key(
key_id: uuid.UUID,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> None:
user = await _get_user(session, user_id)
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()
+2
View File
@@ -3,10 +3,12 @@ from fastapi.staticfiles import StaticFiles
from src.api.auth import router as auth_router
from src.api.projects import router as projects_router
from src.api.ssh_keys import router as ssh_keys_router
from src.api.users import router as users_router
app = FastAPI(title="Headquarter API")
app.include_router(auth_router)
app.include_router(projects_router)
app.include_router(users_router)
app.include_router(ssh_keys_router)
app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads")
+22
View File
@@ -0,0 +1,22 @@
import pytest
from httpx import AsyncClient
from src.main import app
@pytest.fixture
async def async_client():
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
@pytest.mark.asyncio
async def test_create_ssh_key_requires_authentication(async_client: AsyncClient) -> None:
response = await async_client.post("/ssh-keys", json={"name": "test-key"})
assert response.status_code == 401
@pytest.mark.asyncio
async def test_list_ssh_keys_requires_authentication(async_client: AsyncClient) -> None:
response = await async_client.get("/ssh-keys")
assert response.status_code == 401