bruno
5a124d1328
- OAuth2 Gitea (login/callback/logout) with fallback admin mode
- Session management with signed cookies (itsdangerous, 7 days)
- CSRF protection middleware on all POST/PUT/PATCH/DELETE
- Rate limiting in-memory (60 req/min per IP)
- Issue creation from FlowDeck (POST /api/issues/{owner}/{repo})
- Inline issue editing (PATCH /api/issues/{owner}/{repo}/{issue_id})
- Card detail modal (double-click): description, labels, comments
- Checklists with toggleable items (CRUD endpoints + UI)
- Priority + due date editable on cards
- Webhook receiver (POST /api/webhook) with HMAC-SHA256
- Auto-register/status webhooks for repos
- GiteaClient: create_issue, update_issue, get_comments, webhooks, collaborators
- New DB tables: users, user_tokens, checklists, checklist_items
- CSS: modal, forms, checklists, comments, PR section
69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
"""FlowDeck — Session management with signed cookies."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
|
|
from itsdangerous import URLSafeTimedSerializer, BadSignature, SignatureExpired
|
|
|
|
from app.config import settings
|
|
|
|
_serializer = URLSafeTimedSerializer(settings.app_secret_key)
|
|
|
|
|
|
class SessionManager:
|
|
"""Manages user sessions via signed cookies."""
|
|
|
|
@staticmethod
|
|
def create_session(user_data: dict) -> str:
|
|
"""Create a signed session cookie value."""
|
|
payload = {
|
|
"user": user_data,
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
}
|
|
return _serializer.dumps(payload)
|
|
|
|
@staticmethod
|
|
def decode_session(cookie: str) -> dict | None:
|
|
"""Decode and validate a session cookie. Returns user data or None."""
|
|
try:
|
|
payload = _serializer.loads(cookie, max_age=86400 * 7) # 7 days
|
|
return payload.get("user")
|
|
except (BadSignature, SignatureExpired):
|
|
return None
|
|
|
|
@staticmethod
|
|
def store_token(user_id: int, gitea_token: str) -> None:
|
|
"""Store a user's Gitea OAuth token in SQLite."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO user_tokens (gitea_user_id, gitea_token, updated_at)
|
|
VALUES (?, ?, CURRENT_TIMESTAMP)
|
|
ON CONFLICT(gitea_user_id)
|
|
DO UPDATE SET gitea_token=excluded.gitea_token, updated_at=CURRENT_TIMESTAMP""",
|
|
(user_id, gitea_token),
|
|
)
|
|
conn.commit()
|
|
|
|
@staticmethod
|
|
def get_token(user_id: int) -> str | None:
|
|
"""Get a user's stored Gitea token."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT gitea_token FROM user_tokens WHERE gitea_user_id=?",
|
|
(user_id,),
|
|
).fetchone()
|
|
return row["gitea_token"] if row else None
|
|
|
|
|
|
# FastAPI dependency
|
|
async def get_current_user(request) -> dict | None:
|
|
"""FastAPI dependency: extract current user from session cookie."""
|
|
from fastapi import Request
|
|
session = request.cookies.get("flowdeck_session")
|
|
if session:
|
|
return SessionManager.decode_session(session)
|
|
return None
|