176 lines
6.2 KiB
Python
176 lines
6.2 KiB
Python
import re
|
||
from datetime import datetime, timezone
|
||
from urllib.parse import urlparse
|
||
from uuid import UUID
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, status
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy import select
|
||
from sqlalchemy.exc import IntegrityError
|
||
from sqlalchemy.orm import Session
|
||
|
||
from .auth import current_user
|
||
from .database import get_db
|
||
from .models import UrmiOtpProject, UserAccount
|
||
from .otp_service import GenerateOtpPayload, generate_otp_for_user
|
||
from .urmiotp_catalog import OTP_API_DOCUMENTATION, render_templates_for_project
|
||
|
||
router = APIRouter(prefix="/api/v1/urmiotp", tags=["urmiotp"])
|
||
|
||
PROJECT_ID_PATTERN = re.compile(r"^[a-z][a-z0-9-]{4,28}[a-z0-9]$")
|
||
MESSAGE_LABEL_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 -]{0,18}[A-Za-z0-9]$|^[A-Za-z0-9]$")
|
||
RESERVED_TERMS = ("google", "goog", "gcp", "ssl")
|
||
|
||
|
||
class CreateUrmiOtpProjectRequest(BaseModel):
|
||
project_name: str = Field(min_length=6, max_length=30)
|
||
message_label: str = Field(min_length=1, max_length=20)
|
||
app_type: str = Field(pattern="^(web|mobile)$")
|
||
app_url: str = Field(min_length=8, max_length=1000)
|
||
purpose: str = Field(min_length=20, max_length=1000)
|
||
terms_accepted: bool
|
||
|
||
|
||
def validate_project_name(value: str) -> str:
|
||
project_name = value.strip().lower()
|
||
if not PROJECT_ID_PATTERN.fullmatch(project_name):
|
||
raise ValueError(
|
||
"Project name must be 6–30 lowercase letters, numbers, or hyphens; "
|
||
"start with a letter and end with a letter or number"
|
||
)
|
||
if "--" in project_name:
|
||
raise ValueError("Project name cannot contain consecutive hyphens")
|
||
if any(term in project_name for term in RESERVED_TERMS):
|
||
raise ValueError("Project name contains a reserved term")
|
||
return project_name
|
||
|
||
|
||
def validate_message_label(value: str) -> str:
|
||
message_label = value.strip()
|
||
if not message_label or len(message_label) > 20:
|
||
raise ValueError("Message label must be 1–20 characters")
|
||
if not MESSAGE_LABEL_PATTERN.fullmatch(message_label):
|
||
raise ValueError(
|
||
"Message label may only contain letters, numbers, spaces, and hyphens; "
|
||
"start and end with a letter or number"
|
||
)
|
||
return message_label
|
||
|
||
|
||
def validate_app_url(value: str, app_type: str) -> str:
|
||
app_url = value.strip()
|
||
parsed = urlparse(app_url)
|
||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||
raise ValueError("Enter a valid HTTPS website, domain, or app-store URL")
|
||
if app_type == "mobile" and not any(
|
||
host in parsed.netloc.lower()
|
||
for host in ("play.google.com", "apps.apple.com", "appgallery.huawei.com")
|
||
):
|
||
raise ValueError("Mobile apps require a Google Play, Apple App Store, or AppGallery URL")
|
||
return app_url
|
||
|
||
|
||
def serialize_project(project: UrmiOtpProject) -> dict:
|
||
return {
|
||
"id": str(project.id),
|
||
"project_name": project.project_name,
|
||
"message_label": project.message_label,
|
||
"app_type": project.app_type,
|
||
"app_url": project.app_url,
|
||
"purpose": project.purpose,
|
||
"status": project.status,
|
||
"review_note": project.review_note,
|
||
"terms_accepted_at": project.terms_accepted_at,
|
||
"created_at": project.created_at,
|
||
"reviewed_at": project.reviewed_at,
|
||
}
|
||
|
||
|
||
@router.get("/projects")
|
||
def list_projects(
|
||
db: Session = Depends(get_db),
|
||
user: UserAccount = Depends(current_user),
|
||
) -> list[dict]:
|
||
projects = db.scalars(
|
||
select(UrmiOtpProject)
|
||
.where(UrmiOtpProject.user_id == user.id)
|
||
.order_by(UrmiOtpProject.created_at.desc())
|
||
).all()
|
||
return [serialize_project(project) for project in projects]
|
||
|
||
|
||
@router.post("/projects", status_code=status.HTTP_201_CREATED)
|
||
def create_project(
|
||
payload: CreateUrmiOtpProjectRequest,
|
||
db: Session = Depends(get_db),
|
||
user: UserAccount = Depends(current_user),
|
||
) -> dict:
|
||
if not payload.terms_accepted:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||
detail="You must agree to the UrmiOTP terms and conditions",
|
||
)
|
||
try:
|
||
project_name = validate_project_name(payload.project_name)
|
||
message_label = validate_message_label(payload.message_label)
|
||
app_url = validate_app_url(payload.app_url, payload.app_type)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
|
||
|
||
project = UrmiOtpProject(
|
||
user_id=user.id,
|
||
project_name=project_name,
|
||
message_label=message_label,
|
||
app_type=payload.app_type,
|
||
app_url=app_url,
|
||
purpose=payload.purpose.strip(),
|
||
terms_accepted_at=datetime.now(timezone.utc),
|
||
status="pending",
|
||
)
|
||
db.add(project)
|
||
try:
|
||
db.commit()
|
||
except IntegrityError as exc:
|
||
db.rollback()
|
||
raise HTTPException(
|
||
status_code=status.HTTP_409_CONFLICT,
|
||
detail="This project name is already in use",
|
||
) from exc
|
||
db.refresh(project)
|
||
return serialize_project(project)
|
||
|
||
|
||
@router.get("/projects/{project_id}/documentation")
|
||
def project_documentation(
|
||
project_id: UUID,
|
||
db: Session = Depends(get_db),
|
||
user: UserAccount = Depends(current_user),
|
||
) -> dict:
|
||
project = db.scalar(
|
||
select(UrmiOtpProject).where(
|
||
UrmiOtpProject.id == project_id,
|
||
UrmiOtpProject.user_id == user.id,
|
||
)
|
||
)
|
||
if not project:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="UrmiOTP project not found")
|
||
if project.status != "approved":
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail="API documentation is available after project approval",
|
||
)
|
||
return {
|
||
"project": serialize_project(project),
|
||
"templates": render_templates_for_project(project.message_label),
|
||
"api": OTP_API_DOCUMENTATION,
|
||
}
|
||
|
||
|
||
@router.post("/generate")
|
||
def generate_otp_console(
|
||
payload: GenerateOtpPayload,
|
||
db: Session = Depends(get_db),
|
||
user: UserAccount = Depends(current_user),
|
||
) -> dict:
|
||
"""Console/session OTP generate — writes to otp_message_logs for Logs & Analytics."""
|
||
return generate_otp_for_user(db, user, payload)
|