feat: token auth on generate endpoint and regenerate-token endpoint
This commit is contained in:
@@ -1,11 +1,13 @@
|
||||
from typing import Optional
|
||||
from fastapi import APIRouter, Depends, HTTPException, Response
|
||||
from fastapi.responses import StreamingResponse
|
||||
from sqlalchemy.orm import Session, selectinload
|
||||
from app import models, schemas
|
||||
from app.auth import get_current_user
|
||||
from app.auth import get_current_user, get_optional_user
|
||||
from app.database import get_db
|
||||
from app.shorewall_generator import ShorewallGenerator
|
||||
import io
|
||||
import secrets
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@@ -80,9 +82,35 @@ def delete_config(
|
||||
def generate_config(
|
||||
config_id: int,
|
||||
format: str = "json",
|
||||
body: Optional[schemas.GenerateRequest] = None,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: models.User = Depends(get_current_user),
|
||||
current_user: Optional[models.User] = Depends(get_optional_user),
|
||||
):
|
||||
token = body.token if body else None
|
||||
|
||||
# Determine access: OIDC session or matching download token
|
||||
if current_user is not None:
|
||||
# Authenticated via OIDC — enforce owner check
|
||||
config = (
|
||||
db.query(models.Config)
|
||||
.filter(models.Config.id == config_id, models.Config.owner_id == current_user.id)
|
||||
.first()
|
||||
)
|
||||
if not config:
|
||||
raise HTTPException(status_code=404, detail="Config not found")
|
||||
elif token:
|
||||
# Token auth — no owner filter, just match token
|
||||
config = (
|
||||
db.query(models.Config)
|
||||
.filter(models.Config.id == config_id, models.Config.download_token == token)
|
||||
.first()
|
||||
)
|
||||
if not config:
|
||||
raise HTTPException(status_code=401, detail="Invalid token")
|
||||
else:
|
||||
raise HTTPException(status_code=401, detail="Authentication required")
|
||||
|
||||
# Eagerly load relationships
|
||||
config = (
|
||||
db.query(models.Config)
|
||||
.options(
|
||||
@@ -96,11 +124,9 @@ def generate_config(
|
||||
selectinload(models.Config.host_entries).selectinload(models.Host.zone),
|
||||
selectinload(models.Config.params),
|
||||
)
|
||||
.filter(models.Config.id == config_id, models.Config.owner_id == current_user.id)
|
||||
.filter(models.Config.id == config_id)
|
||||
.first()
|
||||
)
|
||||
if not config:
|
||||
raise HTTPException(status_code=404, detail="Config not found")
|
||||
|
||||
generator = ShorewallGenerator(config)
|
||||
|
||||
@@ -113,3 +139,16 @@ def generate_config(
|
||||
)
|
||||
|
||||
return generator.as_json()
|
||||
|
||||
|
||||
@router.post("/{config_id}/regenerate-token", response_model=schemas.RegenerateTokenOut)
|
||||
def regenerate_token(
|
||||
config_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: models.User = Depends(get_current_user),
|
||||
) -> dict:
|
||||
config = _get_config_or_404(config_id, db, current_user)
|
||||
config.download_token = secrets.token_urlsafe(32)
|
||||
db.commit()
|
||||
db.refresh(config)
|
||||
return {"download_token": config.download_token}
|
||||
|
||||
Reference in New Issue
Block a user