feat(users): passwordless user creation via emailed set-password link, display full name instead of username
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled

Admins now create users with a name + email (no password) and role;
a Send Email action (per-user, also reusable for resets) issues a
one-time token and POSTs it to an admin-configurable webhook URL —
intended for a Power Automate flow that delivers the actual email.
The old admin-sets-the-password flow is kept server-side
(LegacyUserCreateWithPassword) but no longer wired into the UI.

Every user-facing surface (top bar, sidebar, user management, audit
log, assignee pickers, dispute notifications) now shows full_name
instead of username, falling back to username when unset.

Also fixes long attack_procedure/expected_detection/suggested_text
text overflowing its container in the template review panels and
Red/Blue team fields (missing break-words on whitespace-pre-wrap
blocks).
This commit is contained in:
kitos
2026-07-17 16:58:25 +02:00
parent 60ab2e558f
commit 4dea19cae9
36 changed files with 974 additions and 159 deletions
@@ -0,0 +1,43 @@
"""Add users.full_name and a password_setup_tokens table.
Users are now created with a name + email instead of a directly-set
password — the admin's "Send Email" action issues a one-time token (this
table) for the user to set their own password via a webhook-delivered
link. The same mechanism is reused for password resets.
Revision ID: b065
Revises: b064
Create Date: 2026-07-17
"""
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
revision = "b065"
down_revision = "b064"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column("users", sa.Column("full_name", sa.String(), nullable=True))
op.create_table(
"password_setup_tokens",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False),
sa.Column("token", sa.String(128), nullable=False),
sa.Column("expires_at", sa.DateTime(), nullable=False),
sa.Column("used_at", sa.DateTime(), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
)
op.create_index("ix_password_setup_tokens_token", "password_setup_tokens", ["token"], unique=True)
op.create_index("ix_password_setup_tokens_user_id", "password_setup_tokens", ["user_id"])
def downgrade() -> None:
op.drop_index("ix_password_setup_tokens_user_id", table_name="password_setup_tokens")
op.drop_index("ix_password_setup_tokens_token", table_name="password_setup_tokens")
op.drop_table("password_setup_tokens")
op.drop_column("users", "full_name")
+2
View File
@@ -48,6 +48,7 @@ from app.models.test_template import TestTemplate
from app.models.procedure_suggestion import ProcedureSuggestion
from app.models.template_suggestion import TemplateSuggestion
from app.models.user import User
from app.models.password_setup_token import PasswordSetupToken
# Assign __all__ = [
__all__ = [
@@ -92,4 +93,5 @@ __all__ = [
"TestRoundHistory",
"ProcedureSuggestion",
"TemplateSuggestion",
"PasswordSetupToken",
]
@@ -0,0 +1,30 @@
"""SQLAlchemy model for one-time password setup/reset tokens.
Issued when an admin sends a "set your password" email — a passwordless
user (freshly created, or one whose password an admin wants to reset)
uses the token to pick their own password without ever having a temporary
one shared with them.
"""
import uuid
from sqlalchemy import Column, DateTime, ForeignKey, String, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from app.database import Base
class PasswordSetupToken(Base):
"""A one-time token letting its owning user set their own password."""
__tablename__ = "password_setup_tokens"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
token = Column(String(128), unique=True, nullable=False, index=True)
expires_at = Column(DateTime, nullable=False)
used_at = Column(DateTime, nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
user = relationship("User", foreign_keys=[user_id])
+3
View File
@@ -30,6 +30,9 @@ class User(Base):
username = Column(String, unique=True, nullable=False)
# Assign email = Column(String, nullable=True)
email = Column(String, nullable=True)
# Display name shown everywhere in the UI instead of username (which is
# now just an internal login identifier, auto-set to the user's email).
full_name = Column(String, nullable=True)
# Assign hashed_password = Column(String, nullable=False)
hashed_password = Column(String, nullable=False)
# Assign role = Column(String, nullable=False, default="viewer")
+41 -1
View File
@@ -57,7 +57,7 @@ from app.models.user import User
from app.schemas.auth import TokenResponse, UserOut
# Import PasswordChange from app.schemas.user
from app.schemas.user import PasswordChange
from app.schemas.user import PasswordChange, SetPasswordRequest, SetPasswordTokenValidateOut
# Import log_action from app.services.audit_service
from app.services.audit_service import log_action
@@ -72,6 +72,12 @@ from app.services.auth_service import (
change_password as auth_change_password,
)
from app.domain.errors import EntityNotFoundError
from app.services.password_setup_service import (
consume_token_and_set_password,
validate_token as validate_password_setup_token,
)
# Assign router = APIRouter(prefix="/auth", tags=["auth"])
router = APIRouter(prefix="/auth", tags=["auth"])
@@ -381,3 +387,37 @@ def change_password(
# Return {"detail": "Password changed successfully"}
return {"detail": "Password changed successfully"}
# ---------------------------------------------------------------------------
# Set password via a one-time emailed link — public (no auth), rate-limited
# ---------------------------------------------------------------------------
@router.get("/set-password/validate", response_model=SetPasswordTokenValidateOut)
@limiter.limit("20/minute")
def validate_set_password_token(request: Request, token: str, db: Session = Depends(get_db)) -> SetPasswordTokenValidateOut:
"""Check a set-password token before rendering the form. Public endpoint."""
try:
user = validate_password_setup_token(db, token)
except (EntityNotFoundError, BusinessRuleViolation):
return SetPasswordTokenValidateOut(valid=False)
return SetPasswordTokenValidateOut(valid=True, full_name=user.full_name)
@router.post("/set-password")
@limiter.limit("10/minute")
def set_password(request: Request, body: SetPasswordRequest, db: Session = Depends(get_db)) -> dict:
"""Complete a password setup/reset using a one-time token. Public endpoint."""
with UnitOfWork(db) as uow:
user = consume_token_and_set_password(db, body.token, body.new_password)
log_action(
db,
user.id,
"SET_PASSWORD_VIA_TOKEN",
"auth",
str(user.id),
details={},
)
uow.commit()
return {"detail": "Password set successfully — you can now log in"}
+42
View File
@@ -344,6 +344,48 @@ def update_jira_config(
)
class PasswordWebhookConfigOut(BaseModel):
configured: bool
url: str
class PasswordWebhookConfigUpdate(BaseModel):
url: str
@router.get("/password-webhook-config", response_model=PasswordWebhookConfigOut)
def get_password_webhook_config(
db: Session = Depends(get_db),
current_user: User = Depends(require_role("admin")),
):
"""Return the configured webhook URL used to email set-password links.
**Requires** the ``admin`` role.
"""
from app.services.password_setup_service import get_password_webhook_url
url = get_password_webhook_url(db) or ""
return PasswordWebhookConfigOut(configured=bool(url), url=url)
@router.patch("/password-webhook-config", response_model=PasswordWebhookConfigOut)
def update_password_webhook_config(
payload: PasswordWebhookConfigUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(require_role("admin")),
):
"""Set the webhook URL used to email set-password links.
**Requires** the ``admin`` role.
"""
from app.services.password_setup_service import get_password_webhook_url, set_password_webhook_url
set_password_webhook_url(db, payload.url)
db.commit()
url = get_password_webhook_url(db) or ""
return PasswordWebhookConfigOut(configured=bool(url), url=url)
@router.post("/jira-test")
def test_jira_connection(
db: Session = Depends(get_db),
+2
View File
@@ -1814,6 +1814,7 @@ def request_discussion(
if rejector_id else None
)
rejector_name = rejector.username if rejector else rejector_label
rejector_full_name = getattr(rejector, "full_name", None) if rejector else None
rejector_email = getattr(rejector, "email", None) if rejector else None
# Notify the rejecting lead
@@ -1852,6 +1853,7 @@ def request_discussion(
"status": "notification_sent",
"message": f"Discussion request sent to {rejector_name}",
"rejector_username": rejector_name,
"rejector_full_name": rejector_full_name,
"rejector_email": rejector_email,
"rejector_role": rejector_label,
}
+49 -12
View File
@@ -21,16 +21,24 @@ from app.domain.unit_of_work import UnitOfWork
# Import User from app.models.user
from app.models.user import User
from app.dependencies.auth import get_current_user
from app.schemas.user import UserCreate, UserUpdate, UserOut, UserPreferencesUpdate, UserOperatorOut
from app.schemas.user import (
UserCreate,
UserUpdate,
UserOut,
UserPreferencesUpdate,
UserOperatorOut,
SendPasswordEmailOut,
)
from app.services.audit_service import log_action
# Import from app.services.user_service
from app.services.user_service import (
create_user,
create_user_without_password,
get_user_or_raise,
list_users,
update_user,
)
from app.services.password_setup_service import send_password_setup_email
# Assign router = APIRouter(prefix="/users", tags=["users"])
router = APIRouter(prefix="/users", tags=["users"])
@@ -110,19 +118,17 @@ def create_user_route(
# Entry: current_user
current_user: User = Depends(require_role("admin")),
) -> UserOut:
"""Create a new user. **Requires admin role.**."""
"""Create a new user. **Requires admin role.**.
No password is set — use ``POST /users/{id}/send-password-email``
afterward to let the user set their own via a one-time link.
"""
# Open context manager
with UnitOfWork(db) as uow:
# Assign user = create_user(
user = create_user(
user = create_user_without_password(
db,
# Keyword argument: username
username=payload.username,
# Keyword argument: email
full_name=payload.full_name,
email=payload.email,
# Keyword argument: password
password=payload.password,
# Keyword argument: role
role=payload.role,
)
# Call log_action()
@@ -137,7 +143,7 @@ def create_user_route(
# Keyword argument: entity_id
entity_id=user.id,
# Keyword argument: details
details={"username": user.username, "role": user.role},
details={"full_name": user.full_name, "email": user.email, "role": user.role},
)
# Call uow.commit()
uow.commit()
@@ -241,3 +247,34 @@ def update_user_route(
# Return user
return user
# ---------------------------------------------------------------------------
# POST /users/{id}/send-password-email — set-password / reset link
# ---------------------------------------------------------------------------
@router.post("/{user_id}/send-password-email", response_model=SendPasswordEmailOut)
def send_password_email_route(
user_id: uuid.UUID,
db: Session = Depends(get_db),
current_user: User = Depends(require_role("admin")),
) -> SendPasswordEmailOut:
"""Email a one-time set-password link to a user. **Requires admin role.**.
Works for both a freshly-created passwordless user and an existing
one whose password needs resetting — same link, same flow either way.
"""
user = get_user_or_raise(db, user_id)
with UnitOfWork(db) as uow:
send_password_setup_email(db, user)
log_action(
db,
user_id=current_user.id,
action="send_password_setup_email",
entity_type="user",
entity_id=user.id,
details={},
)
uow.commit()
return SendPasswordEmailOut(detail=f"Password setup email sent to {user.email}")
+1
View File
@@ -21,6 +21,7 @@ class AuditLogOut(BaseModel):
user_id: uuid.UUID | None = None
# Assign username = None # Populated from user relationship
username: str | None = None # Populated from user relationship
full_name: str | None = None # Populated from user relationship
# action: str
action: str
# Assign entity_type = None
+1
View File
@@ -39,6 +39,7 @@ class UserOut(BaseModel):
id: uuid.UUID
# username: str
username: str
full_name: str | None = None
# Assign email = None
email: str | None = None
# role: str
+74 -33
View File
@@ -112,49 +112,57 @@ def _validate_password_strength(password: str) -> str:
# ── Create ──────────────────────────────────────────────────────────
class UserCreate(BaseModel):
"""Payload for creating a new user."""
"""Payload for creating a new user.
# username: str
username: str
# Assign email = None
email: str | None = None
# password: str
password: str
No password is set here — the admin creates the user with just a
name/email/role, then uses "Send Email" to let the user set their own
password via a one-time link. ``username`` is no longer admin-supplied;
the service layer derives it from ``email`` since it's still needed
internally as the login identifier, but it is never shown in the UI.
"""
full_name: str
email: str
# Assign role = "viewer"
role: str = "viewer"
# Apply the @field_validator decorator
@field_validator("username")
# Apply the @classmethod decorator
@field_validator("full_name")
@classmethod
def full_name_not_blank(cls, v: str) -> str:
v = v.strip()
if not v:
raise ValueError("Full name is required")
return v
@field_validator("email")
@classmethod
def email_format(cls, v: str) -> str:
v = v.strip()
if "@" not in v or v.startswith("@") or v.endswith("@"):
raise ValueError("A valid email address is required")
return v
# ── Legacy direct-password creation (hidden from the UI, kept for a
# possible future revert — see UserCreate above for the active path) ──
class LegacyUserCreateWithPassword(BaseModel):
"""The old admin-sets-the-password creation payload. Not wired to any
active endpoint; retained only so this path can be restored quickly."""
username: str
email: str | None = None
password: str
role: str = "viewer"
@field_validator("username")
@classmethod
# Define function username_format
def username_format(cls, v: str) -> str:
"""Validate the username field against the platform policy.
Args:
v (str): Raw username value from the request body.
Returns:
str: The validated username.
"""
# Return _validate_username(v)
return _validate_username(v)
# Apply the @field_validator decorator
@field_validator("password")
# Apply the @classmethod decorator
@classmethod
# Define function password_strength
def password_strength(cls, v: str) -> str:
"""Validate the password field against the complexity policy.
Args:
v (str): Raw password value from the request body.
Returns:
str: The validated password.
"""
# Return _validate_password_strength(v)
return _validate_password_strength(v)
@@ -163,11 +171,15 @@ class UserCreate(BaseModel):
class UserUpdate(BaseModel):
"""Payload for partially updating an existing user.
Every field is optional so callers send only what changed.
Every field is optional so callers send only what changed. ``password``
is intentionally still supported server-side (see
LegacyUserCreateWithPassword) but the current UI never sends it —
password changes go through the email-a-set-password-link flow instead.
"""
# Assign email = None
email: str | None = None
full_name: str | None = None
# Assign role = None
role: str | None = None
# Assign is_active = None
@@ -248,6 +260,7 @@ class UserOut(BaseModel):
id: uuid.UUID
# username: str
username: str
full_name: str | None = None
# Assign email = None
email: str | None = None
# role: str
@@ -291,6 +304,34 @@ class UserOperatorOut(BaseModel):
id: uuid.UUID
username: str
full_name: str | None = None
role: str
model_config = ConfigDict(from_attributes=True)
# ── Password setup / reset (via emailed one-time token) ─────────────
class SendPasswordEmailOut(BaseModel):
"""Response after an admin triggers a set-password email for a user."""
detail: str
class SetPasswordTokenValidateOut(BaseModel):
"""Response for checking a set-password token before rendering the form."""
valid: bool
full_name: str | None = None
class SetPasswordRequest(BaseModel):
"""Payload for completing a password setup/reset via a one-time token."""
token: str
new_password: str
@field_validator("new_password")
@classmethod
def new_password_strength(cls, v: str) -> str:
return _validate_password_strength(v)
@@ -92,6 +92,7 @@ def list_logs(
"user_id": log.user_id,
# Literal argument value
"username": log.user.username if log.user else None,
"full_name": log.user.full_name if log.user else None,
# Literal argument value
"action": log.action,
# Literal argument value
@@ -0,0 +1,138 @@
"""Passwordless-user onboarding — one-time set-password links sent by webhook.
An admin creates a user with just a name/email/role (see
``user_service.create_user_without_password``). To let that user in, the
admin clicks "Send Email": this service issues a one-time token and POSTs
a payload carrying it to an admin-configured webhook URL (intended for a
Power Automate flow that actually sends the email — the exact webhook
contract is still being finalized, hence the configurable URL rather than
a hardcoded integration). The same mechanism, and the same button, is
reused for password resets on existing users.
"""
from __future__ import annotations
import logging
import secrets
from datetime import datetime, timedelta
import requests
from sqlalchemy.orm import Session
from app.config import settings
from app.domain.errors import BusinessRuleViolation, EntityNotFoundError
from app.models.password_setup_token import PasswordSetupToken
from app.models.user import User
logger = logging.getLogger(__name__)
_WEBHOOK_URL_CONFIG_KEY = "password_setup.webhook_url"
_TOKEN_TTL = timedelta(hours=24)
def _read_system_config(db: Session, key: str) -> str | None:
from app.models.system_config import SystemConfig # avoid circular at import time
row = db.query(SystemConfig).filter(SystemConfig.key == key).first()
return row.value if row else None
def get_password_webhook_url(db: Session) -> str | None:
"""Return the configured Power-Automate-style webhook URL, or None."""
return _read_system_config(db, _WEBHOOK_URL_CONFIG_KEY) or None
def set_password_webhook_url(db: Session, url: str) -> None:
"""Persist the webhook URL. Does not commit; caller commits."""
from app.models.system_config import SystemConfig
row = db.query(SystemConfig).filter(SystemConfig.key == _WEBHOOK_URL_CONFIG_KEY).first()
if row:
row.value = url
else:
db.add(SystemConfig(key=_WEBHOOK_URL_CONFIG_KEY, value=url))
def create_setup_token(db: Session, user: User) -> PasswordSetupToken:
"""Issue a fresh one-time token for *user*, invalidating any unused ones.
Does not commit; caller commits.
"""
db.query(PasswordSetupToken).filter(
PasswordSetupToken.user_id == user.id,
PasswordSetupToken.used_at.is_(None),
).delete()
token = PasswordSetupToken(
user_id=user.id,
token=secrets.token_urlsafe(32),
expires_at=datetime.utcnow() + _TOKEN_TTL,
)
db.add(token)
db.flush()
return token
def send_password_setup_email(db: Session, user: User) -> None:
"""Issue a token and POST it to the configured webhook.
Raises BusinessRuleViolation if no webhook URL is configured yet.
Does not commit; caller commits (the token must be persisted even if
the webhook call itself fails, so a fresh "Send Email" retry works).
"""
webhook_url = get_password_webhook_url(db)
if not webhook_url:
raise BusinessRuleViolation(
"No password-setup webhook is configured yet — set one in Settings first."
)
token = create_setup_token(db, user)
set_password_url = f"{settings.PLATFORM_URL}/set-password?token={token.token}"
try:
requests.post(
webhook_url,
json={
"email": user.email,
"full_name": user.full_name,
"set_password_url": set_password_url,
},
timeout=10,
)
except requests.RequestException:
# Best-effort: the token still exists, so the admin can retry
# "Send Email" once the webhook endpoint is reachable again.
logger.warning("Password-setup webhook call failed for user %s", user.id, exc_info=True)
def _get_valid_token_or_raise(db: Session, token: str) -> PasswordSetupToken:
row = db.query(PasswordSetupToken).filter(PasswordSetupToken.token == token).first()
if row is None:
raise EntityNotFoundError("Password setup token", token)
if row.used_at is not None:
raise BusinessRuleViolation("This link has already been used")
if row.expires_at < datetime.utcnow():
raise BusinessRuleViolation("This link has expired — ask an admin to send a new one")
return row
def validate_token(db: Session, token: str) -> User:
"""Return the token's owning user if the token is valid, else raise."""
row = _get_valid_token_or_raise(db, token)
return row.user
def consume_token_and_set_password(db: Session, token: str, new_password: str) -> User:
"""Set *new_password* for the token's user and mark the token used.
Does not commit; caller commits.
"""
from app.auth import hash_password
row = _get_valid_token_or_raise(db, token)
user = row.user
user.hashed_password = hash_password(new_password)
user.must_change_password = False
row.used_at = datetime.utcnow()
db.flush()
return user
+37
View File
@@ -10,6 +10,9 @@ from __future__ import annotations
# Import uuid
import uuid
# Import secrets
import secrets
# Import Session from sqlalchemy.orm
from sqlalchemy.orm import Session
@@ -88,6 +91,40 @@ def create_user(
return user
def create_user_without_password(db: Session, *, full_name: str, email: str, role: str) -> User:
"""Create a new user without a password — they set their own via an
admin-triggered, emailed one-time link (see password_setup_service).
``username`` is derived from ``email`` since it's still needed
internally as the login identifier, but it's never surfaced in the UI.
Raises DuplicateEntityError if a user with this email/username already
exists. Raises BusinessRuleViolation if role is invalid. Does not
commit; the router handles that.
"""
existing = db.query(User).filter(User.username == email).first()
if existing:
raise DuplicateEntityError("User", "email", email)
if role not in VALID_ROLES:
raise BusinessRuleViolation(
f"Invalid role '{role}'. Must be one of: {', '.join(sorted(VALID_ROLES))}"
)
user = User(
username=email,
email=email,
full_name=full_name,
# Unusable random password — the user can only ever get in via the
# emailed set-password link, never by guessing/being told a value.
hashed_password=hash_password(secrets.token_urlsafe(32)),
role=role,
must_change_password=True,
)
db.add(user)
return user
# Define function get_user_or_raise
def get_user_or_raise(db: Session, user_id: uuid.UUID) -> User:
"""Return a user by ID or raise EntityNotFoundError."""
+132
View File
@@ -0,0 +1,132 @@
"""End-to-end coverage for the passwordless user creation + emailed
set-password-link flow (also reused for password resets).
"""
import uuid
from unittest.mock import patch
import pytest
@pytest.fixture
def new_user_id(api, auth_headers):
resp = api(
"post", "/api/v1/users", auth_headers,
json={"full_name": "Set Password User", "email": "setpw@test.com", "role": "viewer"},
)
assert resp.status_code == 201, resp.text
return resp.json()["id"]
def test_send_password_email_requires_webhook_configured(api, auth_headers, new_user_id):
resp = api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
assert resp.status_code == 400
assert "webhook" in resp.text.lower()
def test_admin_can_configure_password_webhook(api, auth_headers):
resp = api(
"patch", "/api/v1/system/password-webhook-config", auth_headers,
json={"url": "https://example.com/power-automate-hook"},
)
assert resp.status_code == 200, resp.text
assert resp.json()["configured"] is True
get_resp = api("get", "/api/v1/system/password-webhook-config", auth_headers)
assert get_resp.json()["url"] == "https://example.com/power-automate-hook"
def test_send_password_email_posts_to_webhook_and_issues_token(api, db, auth_headers, new_user_id):
api(
"patch", "/api/v1/system/password-webhook-config", auth_headers,
json={"url": "https://example.com/power-automate-hook"},
)
with patch("app.services.password_setup_service.requests.post") as mock_post:
resp = api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
assert resp.status_code == 200, resp.text
mock_post.assert_called_once()
call_kwargs = mock_post.call_args
assert call_kwargs.args[0] == "https://example.com/power-automate-hook"
assert call_kwargs.kwargs["json"]["email"] == "setpw@test.com"
assert "token=" in call_kwargs.kwargs["json"]["set_password_url"]
from app.models.password_setup_token import PasswordSetupToken
token_row = db.query(PasswordSetupToken).filter(PasswordSetupToken.user_id == uuid.UUID(new_user_id)).first()
assert token_row is not None
assert token_row.used_at is None
def test_set_password_with_valid_token_succeeds(api, db, auth_headers, new_user_id):
api(
"patch", "/api/v1/system/password-webhook-config", auth_headers,
json={"url": "https://example.com/power-automate-hook"},
)
with patch("app.services.password_setup_service.requests.post"):
api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
from app.models.password_setup_token import PasswordSetupToken
token_row = db.query(PasswordSetupToken).filter(PasswordSetupToken.user_id == uuid.UUID(new_user_id)).first()
validate = api("get", f"/api/v1/auth/set-password/validate?token={token_row.token}", {})
assert validate.status_code == 200
assert validate.json()["valid"] is True
assert validate.json()["full_name"] == "Set Password User"
resp = api(
"post", "/api/v1/auth/set-password", {},
json={"token": token_row.token, "new_password": "BrandNewPass123!@#"},
)
assert resp.status_code == 200, resp.text
login = api(
"post", "/api/v1/auth/login", {},
data={"username": "setpw@test.com", "password": "BrandNewPass123!@#"},
)
assert login.status_code == 200, login.text
def test_set_password_token_cannot_be_reused(api, db, auth_headers, new_user_id):
api(
"patch", "/api/v1/system/password-webhook-config", auth_headers,
json={"url": "https://example.com/power-automate-hook"},
)
with patch("app.services.password_setup_service.requests.post"):
api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
from app.models.password_setup_token import PasswordSetupToken
token_row = db.query(PasswordSetupToken).filter(PasswordSetupToken.user_id == uuid.UUID(new_user_id)).first()
first = api(
"post", "/api/v1/auth/set-password", {},
json={"token": token_row.token, "new_password": "BrandNewPass123!@#"},
)
assert first.status_code == 200
second = api(
"post", "/api/v1/auth/set-password", {},
json={"token": token_row.token, "new_password": "AnotherPass456!@#"},
)
assert second.status_code == 400
def test_set_password_invalid_token_rejected(api):
validate = api("get", "/api/v1/auth/set-password/validate?token=not-a-real-token", {})
assert validate.status_code == 200
assert validate.json()["valid"] is False
resp = api(
"post", "/api/v1/auth/set-password", {},
json={"token": "not-a-real-token", "new_password": "BrandNewPass123!@#"},
)
assert resp.status_code == 404
def test_password_webhook_config_requires_admin(api, red_lead_headers):
resp = api("get", "/api/v1/system/password-webhook-config", red_lead_headers)
assert resp.status_code == 403
def test_send_password_email_requires_admin(api, red_lead_headers, new_user_id):
resp = api("post", f"/api/v1/users/{new_user_id}/send-password-email", red_lead_headers)
assert resp.status_code == 403
+1 -1
View File
@@ -10,7 +10,7 @@ if backend_dir not in sys.path:
import pytest
from pydantic import ValidationError
from app.schemas.user import UserCreate
from app.schemas.user import LegacyUserCreateWithPassword as UserCreate
# ── Username validation ──────────────────────────────────────────────
+37 -32
View File
@@ -1,28 +1,30 @@
"""API-level validation tests for user creation (SEC-004, SEC-007)."""
"""API-level validation tests for user creation (SEC-004, SEC-007).
Users are created with a name + email no password (see
password_setup_service for the emailed set-password link flow) and no
admin-supplied username (username is auto-derived from email).
"""
def test_create_user_weak_password_rejected(client, admin_user, admin_token):
def test_create_user_missing_full_name_rejected(client, admin_user, admin_token):
response = client.post(
"/api/v1/users",
json={
"username": "newuser",
"password": "123",
"full_name": "",
"email": "new@test.com",
"role": "viewer",
},
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 422
assert "password" in response.text.lower()
def test_create_user_reserved_username(client, admin_user, admin_token):
def test_create_user_invalid_email_rejected(client, admin_user, admin_token):
response = client.post(
"/api/v1/users",
json={
"username": "system",
"password": "SecurePass123!@#",
"email": "sys@test.com",
"full_name": "New User",
"email": "not-an-email",
"role": "viewer",
},
headers={"Authorization": f"Bearer {admin_token}"},
@@ -30,33 +32,21 @@ def test_create_user_reserved_username(client, admin_user, admin_token):
assert response.status_code == 422
def test_create_user_invalid_username_chars(client, admin_user, admin_token):
def test_create_user_valid_request_accepted(client, admin_user, admin_token):
response = client.post(
"/api/v1/users",
json={
"username": "../admin",
"password": "SecurePass123!@#",
"email": "bad@test.com",
"role": "viewer",
},
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 422
def test_create_user_valid_password_accepted(client, admin_user, admin_token):
response = client.post(
"/api/v1/users",
json={
"username": "validuser99",
"password": "ValidPass123!@#",
"full_name": "Valid User",
"email": "valid@test.com",
"role": "viewer",
},
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 201
assert response.json()["username"] == "validuser99"
assert response.status_code == 201, response.text
body = response.json()
assert body["full_name"] == "Valid User"
assert body["username"] == "valid@test.com"
assert body["must_change_password"] is True
def test_create_user_with_manager_role_accepted(client, admin_user, admin_token):
@@ -65,8 +55,7 @@ def test_create_user_with_manager_role_accepted(client, admin_user, admin_token)
response = client.post(
"/api/v1/users",
json={
"username": "newmanager",
"password": "ValidPass123!@#",
"full_name": "New Manager",
"email": "newmanager@test.com",
"role": "manager",
},
@@ -80,8 +69,7 @@ def test_create_user_invalid_role_rejected(client, admin_user, admin_token):
response = client.post(
"/api/v1/users",
json={
"username": "badroleuser",
"password": "ValidPass123!@#",
"full_name": "Bad Role User",
"email": "badrole@test.com",
"role": "superuser",
},
@@ -89,3 +77,20 @@ def test_create_user_invalid_role_rejected(client, admin_user, admin_token):
)
assert response.status_code == 400
assert "Invalid role" in response.text
def test_create_user_duplicate_email_rejected(client, admin_user, admin_token):
payload = {
"full_name": "Dup User",
"email": "dup@test.com",
"role": "viewer",
}
first = client.post(
"/api/v1/users", json=payload, headers={"Authorization": f"Bearer {admin_token}"},
)
assert first.status_code == 201, first.text
second = client.post(
"/api/v1/users", json=payload, headers={"Authorization": f"Bearer {admin_token}"},
)
assert second.status_code == 409