dwindown/yellow-bank-soal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden auth and persist report schedules

Browse Source
This commit is contained in:
dwindown
2026-06-06 19:40:32 +07:00
parent aaf64264f7
commit fd7989f673
18 changed files with 748 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
app/routers/admin.py
View File

@@ -5,12 +5,13 @@ Provides admin-specific endpoints for triggering calibration,
toggling AI generation, and resetting normalization.
"""
from typing import Any, Dict, Optional
from typing import Any, Dict
from fastapi import APIRouter, Depends, Header, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.auth import AuthContext, get_auth_context, require_website_auth
from app.core.config import get_settings
from app.database import get_db
from app.models import Tryout, TryoutStats
@@ -23,35 +24,6 @@ router = APIRouter(prefix="/admin", tags=["admin"])
settings = get_settings()
def get_admin_website_id(
x_website_id: Optional[str] = Header(None, alias="X-Website-ID"),
) -> int:
"""
Extract and validate website_id from request header for admin operations.
Args:
x_website_id: Website ID from header
Returns:
Validated website ID as integer
Raises:
HTTPException: If header is missing or invalid
"""
if x_website_id is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="X-Website-ID header is required",
)
try:
return int(x_website_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="X-Website-ID must be a valid integer",
)
@router.post(
"/{tryout_id}/calibrate",
summary="Trigger IRT calibration",
@@ -60,7 +32,7 @@ def get_admin_website_id(
async def admin_trigger_calibration(
tryout_id: str,
db: AsyncSession = Depends(get_db),
website_id: int = Depends(get_admin_website_id),
auth: AuthContext = Depends(get_auth_context),
) -> Dict[str, Any]:
"""
Trigger IRT calibration for all items in a tryout.
@@ -79,6 +51,8 @@ async def admin_trigger_calibration(
Raises:
HTTPException: If tryout not found or calibration fails
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
# Verify tryout exists
tryout_result = await db.execute(
select(Tryout).where(
@@ -121,7 +95,7 @@ async def admin_trigger_calibration(
async def admin_toggle_ai_generation(
tryout_id: str,
db: AsyncSession = Depends(get_db),
website_id: int = Depends(get_admin_website_id),
auth: AuthContext = Depends(get_auth_context),
) -> Dict[str, Any]:
"""
Toggle AI generation for a tryout.
@@ -139,6 +113,8 @@ async def admin_toggle_ai_generation(
Raises:
HTTPException: If tryout not found
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
# Get tryout
result = await db.execute(
select(Tryout).where(
@@ -175,7 +151,7 @@ async def admin_toggle_ai_generation(
async def admin_reset_normalization(
tryout_id: str,
db: AsyncSession = Depends(get_db),
website_id: int = Depends(get_admin_website_id),
auth: AuthContext = Depends(get_auth_context),
) -> Dict[str, Any]:
"""
Reset normalization for a tryout.
@@ -193,6 +169,8 @@ async def admin_reset_normalization(
Raises:
HTTPException: If tryout or stats not found
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
# Get tryout stats
stats_result = await db.execute(
select(TryoutStats).where(

8
app/routers/ai.py
View File

@@ -78,7 +78,7 @@ async def generate_preview(
- **ai_model**: OpenRouter model to use (default: qwen/qwen2.5-32b-instruct)
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
enforce_rate_limit(
await enforce_rate_limit(
request_http,
scope="ai.generate_preview",
max_requests=40,
@@ -196,7 +196,7 @@ async def generate_save(
- **ai_model**: AI model used for generation
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
enforce_rate_limit(
await enforce_rate_limit(
request_http,
scope="ai.generate_save",
max_requests=40,
@@ -291,8 +291,8 @@ async def get_stats(
"""
Get AI generation statistics.
"""
require_website_auth(auth, allowed_roles={"admin", "system_admin"})
stats = await get_ai_stats(db)
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
stats = await get_ai_stats(db, website_id=website_id)
return AIStatsResponse(
total_ai_items=stats["total_ai_items"],

8
app/routers/import_export.py
View File

@@ -77,7 +77,7 @@ async def preview_import(
HTTPException: If file format is invalid or parsing fails
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
enforce_rate_limit(
await enforce_rate_limit(
request,
scope="import.preview",
max_requests=30,
@@ -181,7 +181,7 @@ async def import_questions(
HTTPException: If file format is invalid, validation fails, or import fails
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
enforce_rate_limit(
await enforce_rate_limit(
request,
scope="import.questions",
max_requests=20,
@@ -351,7 +351,7 @@ async def preview_tryout_json(
db: AsyncSession = Depends(get_db),
) -> dict:
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
enforce_rate_limit(
await enforce_rate_limit(
request,
scope="import.tryout_json_preview",
max_requests=30,
@@ -394,7 +394,7 @@ async def import_tryout_json(
db: AsyncSession = Depends(get_db),
) -> dict:
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
enforce_rate_limit(
await enforce_rate_limit(
request,
scope="import.tryout_json",
max_requests=20,

33
app/routers/reports.py
View File

@@ -85,6 +85,15 @@ async def get_student_performance_report(
Returns individual student records and/or aggregate statistics.
"""
website_id = require_website_auth(auth, allowed_roles={"student", "admin", "system_admin"})
scoped_wp_user_id = None
if auth.role == "student":
if not auth.wp_user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Student reports require an authenticated WordPress user",
)
scoped_wp_user_id = auth.wp_user_id
date_range = None
if date_start or date_end:
date_range = {}
@@ -99,6 +108,7 @@ async def get_student_performance_report(
db=db,
date_range=date_range,
format_type=format_type,
wp_user_id=scoped_wp_user_id,
)
return _convert_student_performance_report(report)
@@ -361,7 +371,8 @@ async def create_report_schedule(
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
ensure_website_scope_matches(website_id, request.website_id)
schedule_id = schedule_report(
schedule_id = await schedule_report(
db,
report_type=request.report_type,
schedule=request.schedule,
tryout_ids=request.tryout_ids,
@@ -370,7 +381,7 @@ async def create_report_schedule(
export_format=request.export_format,
)
scheduled = get_scheduled_report(schedule_id)
scheduled = await get_scheduled_report(db, schedule_id)
return ReportScheduleResponse(
schedule_id=schedule_id,
@@ -387,6 +398,7 @@ async def create_report_schedule(
)
async def get_scheduled_report_details(
schedule_id: str,
db: AsyncSession = Depends(get_db),
auth: AuthContext = Depends(get_auth_context),
) -> ReportScheduleOutput:
"""
@@ -395,7 +407,7 @@ async def get_scheduled_report_details(
Returns the configuration and status of a scheduled report.
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
scheduled = get_scheduled_report(schedule_id)
scheduled = await get_scheduled_report(db, schedule_id)
if not scheduled:
raise HTTPException(
@@ -431,6 +443,7 @@ async def get_scheduled_report_details(
description="List all scheduled reports for a website.",
)
async def list_scheduled_reports_endpoint(
db: AsyncSession = Depends(get_db),
auth: AuthContext = Depends(get_auth_context),
) -> List[ReportScheduleOutput]:
"""
@@ -439,7 +452,7 @@ async def list_scheduled_reports_endpoint(
Returns all scheduled reports for the current website.
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
reports = list_scheduled_reports(website_id=website_id)
reports = await list_scheduled_reports(db, website_id=website_id)
return [
ReportScheduleOutput(
@@ -466,6 +479,7 @@ async def list_scheduled_reports_endpoint(
)
async def cancel_scheduled_report_endpoint(
schedule_id: str,
db: AsyncSession = Depends(get_db),
auth: AuthContext = Depends(get_auth_context),
) -> dict:
"""
@@ -474,7 +488,7 @@ async def cancel_scheduled_report_endpoint(
Removes the scheduled report from the system.
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
scheduled = get_scheduled_report(schedule_id)
scheduled = await get_scheduled_report(db, schedule_id)
if not scheduled:
raise HTTPException(
@@ -488,7 +502,7 @@ async def cancel_scheduled_report_endpoint(
detail="Access denied to this scheduled report",
)
success = cancel_scheduled_report(schedule_id)
success = await cancel_scheduled_report(db, schedule_id)
if not success:
raise HTTPException(
@@ -523,7 +537,7 @@ async def export_scheduled_report(
Generates the report and returns it as a file download.
"""
website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
scheduled = get_scheduled_report(schedule_id)
scheduled = await get_scheduled_report(db, schedule_id)
if not scheduled:
raise HTTPException(
@@ -536,6 +550,11 @@ async def export_scheduled_report(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied to this scheduled report",
)
if not scheduled.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Scheduled report is inactive",
)
# Generate report based on type
report = None

31
app/routers/sessions.py
View File

@@ -10,6 +10,7 @@ Endpoints:
from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
@@ -122,6 +123,27 @@ async def complete_session(
items = {item.id: item for item in items_result.scalars().all()}
# Process each answer
submitted_item_ids = [answer.item_id for answer in request.user_answers]
if len(submitted_item_ids) != len(set(submitted_item_ids)):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Duplicate item answers are not allowed in a session completion",
)
existing_answers_result = await db.execute(
select(UserAnswer.item_id).where(UserAnswer.session_id == session.session_id)
)
existing_answered_item_ids = {row[0] for row in existing_answers_result.all()}
duplicate_existing_ids = sorted(set(submitted_item_ids) & existing_answered_item_ids)
if duplicate_existing_ids:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail={
"message": "One or more items were already answered for this session",
"item_ids": duplicate_existing_ids,
},
)
total_benar = 0
total_bobot_earned = 0.0
user_answer_records = []
@@ -234,7 +256,13 @@ async def complete_session(
await update_tryout_stats(db, website_id, session.tryout_id, nm)
# Commit all changes
await db.commit()
try:
await db.commit()
except IntegrityError as exc:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Duplicate item answer detected for this session",
) from exc
# Refresh to get updated relationships
await db.refresh(session)
@@ -261,7 +289,6 @@ async def complete_session(
id=ua.id,
item_id=ua.item_id,
response=ua.response,
is_correct=ua.is_correct,
time_spent=ua.time_spent,
bobot_earned=ua.bobot_earned,
scoring_mode_used=ua.scoring_mode_used,

39
app/routers/wordpress.py
View File

@@ -15,7 +15,13 @@ from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.core.auth import issue_access_token
from app.core.auth import (
AuthContext,
ensure_website_scope_matches,
get_auth_context,
issue_access_token,
require_website_auth,
)
from app.models.user import User
from app.models.website import Website
from app.schemas.wordpress import (
@@ -44,6 +50,16 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/wordpress", tags=["wordpress"])
def _api_role_from_wordpress_roles(roles: list[str]) -> str:
"""Map WordPress roles to API roles used by route authorization."""
normalized_roles = {str(role).strip().lower() for role in roles}
if normalized_roles & {"super_admin", "system_admin"}:
return "system_admin"
if normalized_roles & {"administrator", "admin"}:
return "admin"
return "student"
def get_website_id_from_header(
x_website_id: Optional[str] = Header(None, alias="X-Website-ID"),
) -> int:
@@ -132,7 +148,7 @@ async def sync_users_endpoint(
Raises:
HTTPException: If website not found, token invalid, or API error
"""
enforce_rate_limit(
await enforce_rate_limit(
request,
scope="wordpress.sync_users",
max_requests=20,
@@ -230,7 +246,7 @@ async def verify_session_endpoint(
Raises:
HTTPException: If website not found or API error
"""
enforce_rate_limit(
await enforce_rate_limit(
http_request,
scope="wordpress.verify_session",
max_requests=60,
@@ -273,7 +289,7 @@ async def verify_session_endpoint(
},
access_token=issue_access_token(
website_id=request.website_id,
role="student",
role=_api_role_from_wordpress_roles(wp_user_info.roles),
wp_user_id=request.wp_user_id,
expires_in_seconds=3600 * 24,
),
@@ -310,6 +326,7 @@ async def verify_session_endpoint(
async def get_website_users(
website_id: int,
db: AsyncSession = Depends(get_db),
auth: AuthContext = Depends(get_auth_context),
page: int = 1,
page_size: int = 50,
) -> UserListResponse:
@@ -328,6 +345,9 @@ async def get_website_users(
Raises:
HTTPException: If website not found
"""
auth_website_id = require_website_auth(auth, allowed_roles={"admin", "system_admin"})
ensure_website_scope_matches(auth_website_id, website_id)
# Validate website exists
await get_valid_website(website_id, db)
@@ -374,6 +394,7 @@ async def get_user_endpoint(
website_id: int,
wp_user_id: str,
db: AsyncSession = Depends(get_db),
auth: AuthContext = Depends(get_auth_context),
) -> WordPressUserResponse:
"""
Get a specific user by WordPress user ID.
@@ -389,6 +410,16 @@ async def get_user_endpoint(
Raises:
HTTPException: If website or user not found
"""
auth_website_id = require_website_auth(
auth, allowed_roles={"student", "admin", "system_admin"}
)
ensure_website_scope_matches(auth_website_id, website_id)
if auth.role == "student" and auth.wp_user_id != wp_user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User does not belong to this authenticated user",
)
# Validate website exists
await get_valid_website(website_id, db)