Add profile feature, input sanitization, and stronger security checks

Backend:
- app/core/sanitize.py: shared sanitize_str, normalize_email, validate_phone,
  validate_date_of_birth — applied to every user-supplied DB-bound input
- app/schemas/user.py: sanitize full_name, normalize email on UserCreate
- app/models/profile.py: profiles table (position, phone, dob, address, updated_at)
- app/models/user.py: Profile back-ref, is_superuser admin-role comment
- app/schemas/profile.py: ProfileRead/ProfileUpdate with full sanitization
- app/routers/profile.py: GET+PUT /api/profile/me (lazy profile creation)
- app/main.py: register /api/profile router
- alembic migration 676084df61d1: create profiles table

Frontend:
- components/Nav.tsx: shared nav (Dashboard | Profile | Logout)
- pages/ProfilePage.tsx: profile view + inline edit form with error handling
- pages/DashboardPage.tsx: use Nav component
- api/client.ts: ProfileData type, getProfile, updateProfile
- App.tsx: /profile private route

Security:
- scripts/security_check.py: tighter SQL injection patterns (f-string/format/%
  in execute/query/text()), new SANIT category for raw request→DB patterns

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
curo1305
2026-04-13 18:15:47 +02:00
parent e117a33a73
commit 343f12259c
18 changed files with 547 additions and 16 deletions
@@ -0,0 +1,40 @@
"""add profiles table
Revision ID: 676084df61d1
Revises: 38efeff7c45a
Create Date: 2026-04-13 16:11:46.705481
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = '676084df61d1'
down_revision: Union[str, None] = '38efeff7c45a'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('profiles',
sa.Column('id', sa.String(), nullable=False),
sa.Column('user_id', sa.String(), nullable=False),
sa.Column('phone', sa.String(length=20), nullable=True),
sa.Column('date_of_birth', sa.Date(), nullable=True),
sa.Column('position', sa.String(length=128), nullable=True),
sa.Column('address', sa.String(length=255), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('user_id')
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('profiles')
# ### end Alembic commands ###
+74
View File
@@ -0,0 +1,74 @@
"""
Input sanitization utilities.
Every string that originates from user input and is destined for the database
MUST pass through these helpers before reaching a SQLAlchemy model or query.
SQLAlchemy's ORM already uses bound parameters (no raw SQL), so these helpers
address the layer above: ensuring data is well-formed, length-capped, and free
of null bytes or control characters before it is stored.
"""
import re
import unicodedata
from datetime import date
# ── Constants ─────────────────────────────────────────────────────────────────
_PHONE_RE = re.compile(r"^\+?[\d\s\-()\[\]]{7,20}$")
# ── Core helper ───────────────────────────────────────────────────────────────
def sanitize_str(value: str | None, max_len: int = 255) -> str | None:
"""Strip whitespace, reject null bytes and non-printable control characters,
enforce a maximum length. Returns None unchanged so optional fields work
naturally with ``Optional[str]`` annotations."""
if value is None:
return None
# Strip leading/trailing whitespace
value = value.strip()
# Reject null bytes (common injection vector)
if "\x00" in value:
raise ValueError("Input must not contain null bytes")
# Reject ASCII control characters (0x010x1F, 0x7F) except tab/newline/CR
# which may appear in multi-line fields. Use Unicode category 'Cc'.
for ch in value:
if unicodedata.category(ch) == "Cc" and ch not in ("\t", "\n", "\r"):
raise ValueError("Input contains invalid control characters")
if len(value) > max_len:
raise ValueError(f"Input must not exceed {max_len} characters")
return value if value != "" else None
def normalize_email(value: str) -> str:
"""Lowercase and strip an email address."""
return value.strip().lower()
def validate_phone(value: str | None) -> str | None:
"""Sanitize then validate phone number format."""
value = sanitize_str(value, max_len=20)
if value is None:
return None
if not _PHONE_RE.match(value):
raise ValueError(
"Phone number may only contain digits, spaces, +, -, (, ) and [ ] "
"and must be 720 characters"
)
return value
def validate_date_of_birth(value: date | None) -> date | None:
"""Reject obviously invalid birth dates (before 1900 or in the future)."""
if value is None:
return None
if value.year < 1900:
raise ValueError("Date of birth must be 1900 or later")
if value > date.today():
raise ValueError("Date of birth must not be in the future")
return value
+2 -1
View File
@@ -2,7 +2,7 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings
from app.routers import auth, users
from app.routers import auth, profile, users
app = FastAPI(title=settings.PROJECT_NAME, version="0.1.0")
@@ -16,6 +16,7 @@ app.add_middleware(
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(users.router, prefix="/api/users", tags=["users"])
app.include_router(profile.router, prefix="/api/profile", tags=["profile"])
@app.get("/api/health")
+2 -1
View File
@@ -1,3 +1,4 @@
from app.models.profile import Profile
from app.models.user import User
__all__ = ["User"]
__all__ = ["User", "Profile"]
+34
View File
@@ -0,0 +1,34 @@
import uuid
from datetime import date, datetime
from sqlalchemy import Date, DateTime, ForeignKey, String, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
class Profile(Base):
__tablename__ = "profiles"
id: Mapped[str] = mapped_column(
String, primary_key=True, default=lambda: str(uuid.uuid4())
)
# One-to-one with users; deleting a user cascades to the profile.
user_id: Mapped[str] = mapped_column(
String, ForeignKey("users.id", ondelete="CASCADE"), unique=True, nullable=False
)
phone: Mapped[str | None] = mapped_column(String(20), nullable=True)
date_of_birth: Mapped[date | None] = mapped_column(Date, nullable=True)
# Job title / role within the organisation (e.g. "Software Engineer", "HR Manager").
position: Mapped[str | None] = mapped_column(String(128), nullable=True)
address: Mapped[str | None] = mapped_column(String(255), nullable=True)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False,
)
user = relationship("User", back_populates="profile")
+11 -1
View File
@@ -1,10 +1,14 @@
import uuid
from typing import TYPE_CHECKING
from sqlalchemy import Boolean, String
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
if TYPE_CHECKING:
from app.models.profile import Profile
class User(Base):
__tablename__ = "users"
@@ -14,4 +18,10 @@ class User(Base):
hashed_password: Mapped[str] = mapped_column(String, nullable=False)
full_name: Mapped[str] = mapped_column(String, nullable=True)
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
# Role flag — True = admin, False = regular user.
# Never exposed in API responses; set only by direct DB or admin tooling.
is_superuser: Mapped[bool] = mapped_column(Boolean, default=False)
profile: Mapped["Profile"] = relationship(
"Profile", back_populates="user", uselist=False, cascade="all, delete-orphan"
)
+48
View File
@@ -0,0 +1,48 @@
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.deps import get_current_user
from app.models.profile import Profile
from app.models.user import User
from app.schemas.profile import ProfileRead, ProfileUpdate
router = APIRouter()
async def _get_or_create_profile(user: User, db: AsyncSession) -> Profile:
"""Return the user's profile, creating an empty one on first access."""
result = await db.execute(select(Profile).where(Profile.user_id == user.id))
profile = result.scalar_one_or_none()
if profile is None:
profile = Profile(user_id=user.id)
db.add(profile)
await db.commit()
await db.refresh(profile)
return profile
@router.get("/me", response_model=ProfileRead)
async def get_my_profile(
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
) -> Profile:
return await _get_or_create_profile(current_user, db)
@router.put("/me", response_model=ProfileRead)
async def update_my_profile(
body: ProfileUpdate,
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
) -> Profile:
profile = await _get_or_create_profile(current_user, db)
# Only update fields that were explicitly provided in the request body.
for field, value in body.model_dump(exclude_unset=True).items():
setattr(profile, field, value)
await db.commit()
await db.refresh(profile)
return profile
+44
View File
@@ -0,0 +1,44 @@
from datetime import date, datetime
from pydantic import BaseModel, field_validator
from app.core.sanitize import sanitize_str, validate_date_of_birth, validate_phone
class ProfileRead(BaseModel):
id: str
user_id: str
phone: str | None
date_of_birth: date | None
position: str | None
address: str | None
updated_at: datetime
model_config = {"from_attributes": True}
class ProfileUpdate(BaseModel):
phone: str | None = None
date_of_birth: date | None = None
position: str | None = None
address: str | None = None
@field_validator("phone", mode="before")
@classmethod
def clean_phone(cls, v: str | None) -> str | None:
return validate_phone(v)
@field_validator("position", mode="before")
@classmethod
def clean_position(cls, v: str | None) -> str | None:
return sanitize_str(v, max_len=128)
@field_validator("address", mode="before")
@classmethod
def clean_address(cls, v: str | None) -> str | None:
return sanitize_str(v, max_len=255)
@field_validator("date_of_birth", mode="after")
@classmethod
def clean_dob(cls, v: date | None) -> date | None:
return validate_date_of_birth(v)
+12
View File
@@ -2,6 +2,8 @@ import re
from pydantic import BaseModel, EmailStr, field_validator
from app.core.sanitize import normalize_email, sanitize_str
# Common words that must not appear as whole words inside a password.
# Checked case-insensitively with word boundaries.
_FORBIDDEN_WORDS = {
@@ -45,6 +47,16 @@ class UserCreate(BaseModel):
password: str
full_name: str | None = None
@field_validator("email", mode="before")
@classmethod
def normalize_email_field(cls, v: str) -> str:
return normalize_email(v)
@field_validator("full_name", mode="before")
@classmethod
def sanitize_full_name(cls, v: str | None) -> str | None:
return sanitize_str(v, max_len=128)
@field_validator("password")
@classmethod
def password_strength(cls, v: str) -> str: