n.siegfried/Project-SDE-WP-Suite
3
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Wave 1: permissions roles, account-backed SOP team, critical constraints, password reset, BIM flag

Browse Source 
Acts on the site comments from 8/3 plus the follow-ups. Foundation work first —
four of the comments all needed the project team to resolve to real user accounts.

Permissions vs project role (new)
- User.role is now the PERMISSIONS role: admin | project_admin | project_user.
  project_admin may delete work packages, change a SOP after it is complete, and
  delete a project; project_user may not (archiving a WP is still open to them).
  Enforced by require_project_admin() server-side; the UI only hides dead ends.
- New User.project_role holds the person's JOB FUNCTION on the project. It grants
  nothing — it feeds the SOP team pickers and notification routing.
- Admin console shows both columns and explains the difference. Migration rewrites
  the legacy role 'user' to 'project_user'.
- Deleting a project was previously open to any member and unaudited; it now needs
  project_admin and writes an audit event. ProjectData.remove no longer drops the
  project from the local cache when the server refuses.

SOP project team from user accounts
- PM/APM/CM/QM and additional team members are pickers over the project's members,
  storing the account id next to the display name. A name from an older SOP with no
  matching account is kept and flagged rather than dropped.
- The WP Creator lists the SOP team first in the Owner picker, and a new package
  defaults to whoever is creating it.

Critical constraints
- SOP constraints carry a Critical flag; buildConstraints() now copies the whole
  definition through to the package (it previously reduced them to names, losing
  description too), and critical rows are marked in the WP form. The email on
  reopen-after-release is wave 3.

Password reset by email
- login.html gains Forgot password and a set-a-new-password view, offered only when
  the server reports email is actually configured.
- Single-use signed token (AUTH_RESET_MINUTES, default 60) bound to token_version,
  sent immediately rather than through the notifications outbox so a reset link is
  never persisted. Identical response for unknown accounts; per-account send
  cooldown; a completed reset clears any login lockout.
- Session and reset tokens are no longer interchangeable.

BIM kill-switch
- New admin Features card with bim_enabled, OFF by default. The SOP creator hides
  the BIM section and the Creator treats every package as install-only while it is
  off; a SOP that already has BIM keeps its data untouched.

Verified with two throwaway-database test scripts: 44 checks on the permissions
matrix and token handling, 22 on the reset flow end-to-end against a local SMTP
sink (real message captured, link extracted and used). Front-end files parse-checked
in headless Chrome. Not yet exercised in a browser against a real login.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Nick Siegfried
2026-08-03 14:48:59 -07:00
parent 1d004cab75
commit 79b0e955b4
16 changed files with 1106 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

218
server/app.py
View File

@@ -12,6 +12,7 @@ import os
import re
import uuid
from datetime import timedelta, timezone
from time import monotonic
from typing import Any, Optional
from urllib.parse import urlparse
@@ -111,7 +112,7 @@ def check_id(v: Optional[str]) -> None:
# mutating endpoints raise 403 on no access.
def accessible_project_ids(db: Session, user: "models.User"):
"""Return the set of project ids the user may access, or None for 'all' (admin)."""
if user.role == "admin":
if auth.is_admin(user):
return None
rows = db.scalars(
select(models.ProjectMember.project_id).where(models.ProjectMember.user_id == user.id)
@@ -120,7 +121,7 @@ def accessible_project_ids(db: Session, user: "models.User"):
def require_project_access(db: Session, user: "models.User", project_id: Optional[str]) -> None:
if user.role == "admin":
if auth.is_admin(user):
return
if not project_id:
# Non-admins may not read/mutate resources with no project assignment
@@ -136,6 +137,19 @@ def require_project_access(db: Session, user: "models.User", project_id: Optiona
raise HTTPException(status_code=403, detail="You don't have access to this project")
def require_project_admin(db: Session, user: "models.User", project_id: Optional[str],
what: str = "this action") -> None:
"""Destructive / baseline-changing operations: deleting a work package or a
project, and editing a SOP that has already been completed. Requires project
access AND the project_admin (or admin) permissions role."""
require_project_access(db, user, project_id)
if not auth.is_project_admin(user):
raise HTTPException(
status_code=403,
detail=f"{what} requires the Project Admin permissions role",
)
def scope_to_access(stmt, column, db: Session, user: "models.User"):
"""Restrict a SELECT to the user's accessible projects (no-op for admins)."""
ids = accessible_project_ids(db, user)
@@ -174,7 +188,7 @@ def require_assignable(db: Session, user_id: str, project_id: Optional[str]) ->
u = db.get(models.User, user_id)
if not u or not u.is_active:
raise HTTPException(status_code=400, detail="Assignee is not a valid user")
if u.role == "admin":
if auth.is_admin(u):
return
ok = db.scalar(
select(models.ProjectMember.id).where(
@@ -251,6 +265,7 @@ class SettingsIn(BaseModel):
from_addr: Optional[str] = None
from_name: Optional[str] = None
app_base_url: Optional[str] = None
bim_enabled: Optional[bool] = None
class TestEmailIn(BaseModel):
@@ -296,7 +311,21 @@ class NewUserIn(BaseModel):
password: str
full_name: str = ""
email: str = ""
role: str = "user" # 'admin' | 'user'
role: str = auth.ROLE_PROJECT_USER # permissions role — see auth.ROLES
project_role: str = "" # job function on the project (no permissions)
class ProjectRoleIn(BaseModel):
project_role: str = ""
class ForgotPasswordIn(BaseModel):
username: str = "" # username or email
class ResetPasswordIn(BaseModel):
token: str
new_password: str
class PasswordChangeIn(BaseModel):
@@ -313,7 +342,7 @@ class ActiveIn(BaseModel):
class RoleIn(BaseModel):
role: str # 'admin' | 'user'
role: str # permissions role — see auth.ROLES
class ProjectAssignIn(BaseModel):
@@ -367,6 +396,115 @@ def logout(response: Response):
return {"ok": True}
# ── Self-service password reset (needs email switched on) ──────────────────────
RESET_COOLDOWN_SECONDS = int(os.getenv("AUTH_RESET_COOLDOWN_SECONDS", "120"))
# In-process throttle: one reset mail per (account, client) per cooldown. Enough to
# stop someone using the form to spam a colleague's inbox. Per-worker and lost on
# restart — deliberately simple; the token expiry is the real control.
_reset_last: dict[str, float] = {}
def _reset_throttled(request: Request, username: str) -> bool:
now = monotonic()
key = f"{(username or '').strip().lower()}|{request.client.host if request.client else ''}"
prev = _reset_last.get(key)
if prev is not None and (now - prev) < RESET_COOLDOWN_SECONDS:
return True
_reset_last[key] = now
if len(_reset_last) > 5000: # bound the dict on a long-lived worker
cutoff = now - RESET_COOLDOWN_SECONDS
for k in [k for k, t in _reset_last.items() if t < cutoff]:
_reset_last.pop(k, None)
return False
def reset_body(user: "models.User", link: str, minutes: int) -> str:
# No account detail beyond the username, and no customer data — same rule as
# the assignment mail. The link is the only sensitive thing in here.
who = user.full_name or user.username
return (
f"Hi {who},\n\n"
f"A password reset was requested for your Work Package Suite account "
f"({user.username}).\n\n"
f"Set a new password:\n{link}\n\n"
f"The link expires in {minutes} minutes and can only be used once. "
f"If you didn't request this, you can ignore this email — your current "
f"password still works.\n"
)
@app.get("/api/auth/reset-available")
def reset_available(db: Session = Depends(get_db)):
"""Whether the login page should offer 'Forgot password'. Self-service reset
depends entirely on outbound email, so it's off unless email is enabled AND
SMTP is configured — otherwise the only route is an admin reset."""
s = notify.get_settings(db)
return {"enabled": bool(s.get("email_enabled")) and notify.smtp_ready(s)}
@app.post("/api/auth/forgot-password")
def forgot_password(body: ForgotPasswordIn, request: Request, db: Session = Depends(get_db)):
"""Email a reset link. Always returns the same 200 response whether or not the
account exists — this endpoint is unauthenticated, so it must not become a
username/email oracle. Failures are recorded in the audit log instead."""
s = notify.get_settings(db)
if not (s.get("email_enabled") and notify.smtp_ready(s)):
raise HTTPException(
status_code=503,
detail="Password reset by email isn't available. Ask an administrator to reset it for you.",
)
if _reset_throttled(request, body.username):
# Same shape as the success response — no oracle, no mail bomb.
return {"ok": True, "message": "If that account exists, a reset link is on its way."}
user = auth.find_user(db, body.username)
if user and user.is_active and user.email:
base = (s.get("app_base_url") or "").rstrip("/")
token = auth.create_reset_token(user)
link = f"{base}/login.html?reset={token}" if base else f"/login.html?reset={token}"
sent = notify.send_now(
db, user.email,
"Work Package Suite — reset your password",
reset_body(user, link, auth.RESET_MINUTES),
)
log_event(db, user.username, "password_reset_requested", "user", user.id,
summary=user.username, detail={"emailed": bool(sent)})
db.commit()
else:
# Log the miss for the admin's benefit; the caller can't tell the difference.
log_event(db, "(anonymous)", "password_reset_miss", "user", "",
summary=(body.username or "")[:200],
detail={"reason": "no account, inactive, or no email on file"})
db.commit()
return {"ok": True, "message": "If that account exists, a reset link is on its way."}
@app.post("/api/auth/reset-password")
def reset_password(body: ResetPasswordIn, db: Session = Depends(get_db)):
"""Complete a reset using the emailed token. The token carries the user's
token_version, and finishing a reset bumps it — so the link is single-use and
every existing session for that account is signed out."""
claims = auth.decode_reset_token(body.token or "")
if not claims:
raise HTTPException(status_code=400, detail="This reset link is invalid or has expired. Request a new one.")
user = db.get(models.User, claims.get("sub"))
if not user or not user.is_active:
raise HTTPException(status_code=400, detail="This reset link is no longer valid.")
if (claims.get("ver", 0) or 0) != (user.token_version or 0):
raise HTTPException(status_code=400, detail="This reset link has already been used. Request a new one.")
problem = auth.password_problem(body.new_password, user.username, user.email)
if problem:
raise HTTPException(status_code=400, detail=problem)
user.password_hash = auth.hash_password(body.new_password)
user.token_version = (user.token_version or 0) + 1 # burns the link + all sessions
# A completed reset also clears any login lockout — the person has proven
# control of the mailbox, so there's nothing left to throttle.
user.failed_attempts = 0
user.locked_until = None
log_event(db, user.username, "password_reset", "user", user.id, summary=user.username)
db.commit()
return {"ok": True}
@app.get("/api/auth/me")
def whoami(user: models.User = Depends(auth.get_current_user)):
"""Who is logged in. The frontend guard calls this on every page load."""
@@ -401,8 +539,8 @@ def create_user(body: NewUserIn, _admin: models.User = Depends(auth.require_admi
problem = auth.password_problem(body.password, body.username, body.email)
if problem:
raise HTTPException(status_code=400, detail=problem)
if body.role not in ("admin", "user"):
raise HTTPException(status_code=400, detail="role must be 'admin' or 'user'")
if body.role not in auth.ROLES:
raise HTTPException(status_code=400, detail=f"role must be one of {', '.join(auth.ROLES)}")
if auth.find_user(db, body.username):
raise HTTPException(status_code=409, detail="A user with that username already exists")
u = models.User(
@@ -412,9 +550,11 @@ def create_user(body: NewUserIn, _admin: models.User = Depends(auth.require_admi
full_name=body.full_name.strip(),
password_hash=auth.hash_password(body.password),
role=body.role,
project_role=body.project_role.strip()[:120],
)
db.add(u)
log_event(db, _admin, "user_created", "user", u.id, summary=u.username, detail={"role": u.role})
log_event(db, _admin, "user_created", "user", u.id, summary=u.username,
detail={"role": u.role, "project_role": u.project_role})
db.commit()
db.refresh(u)
return u.to_dict()
@@ -450,20 +590,22 @@ def set_user_active(user_id: str, body: ActiveIn, admin: models.User = Depends(a
@app.post("/api/auth/users/{user_id}/role")
def set_user_role(user_id: str, body: RoleIn, admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
"""Change a user's role (admin ↔ user). Admins can do this at any time.
"""Change a user's PERMISSIONS role (admin / project_admin / project_user).
Their job function on the project is separate — see set_user_project_role.
Guards: you can't change your own role (avoids self-lockout), and the last
remaining admin can't be demoted (keeps the app manageable)."""
if body.role not in ("admin", "user"):
raise HTTPException(status_code=400, detail="role must be 'admin' or 'user'")
if body.role not in auth.ROLES:
raise HTTPException(status_code=400, detail=f"role must be one of {', '.join(auth.ROLES)}")
u = db.get(models.User, user_id)
if not u:
raise HTTPException(status_code=404, detail="User not found")
if u.id == admin.id:
raise HTTPException(status_code=400, detail="You cannot change your own role")
if u.role == "admin" and body.role != "admin":
if auth.is_admin(u) and body.role != auth.ROLE_ADMIN:
other_admins = db.scalars(
select(models.User.id).where(
(models.User.role == "admin")
(models.User.role == auth.ROLE_ADMIN)
& (models.User.id != u.id)
& (models.User.is_active.is_(True))
)
@@ -479,6 +621,23 @@ def set_user_role(user_id: str, body: RoleIn, admin: models.User = Depends(auth.
return u.to_dict()
@app.post("/api/auth/users/{user_id}/project-role")
def set_user_project_role(user_id: str, body: ProjectRoleIn, admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
"""Set a user's job function on the project (Project Manager, Superintendent,
…). Purely descriptive — it grants nothing. This is what the SOP team pickers
and notification routing read, so it's worth keeping accurate."""
u = db.get(models.User, user_id)
if not u:
raise HTTPException(status_code=404, detail="User not found")
old = u.project_role or ""
u.project_role = (body.project_role or "").strip()[:120]
log_event(db, admin, "project_role_changed", "user", u.id, summary=u.username,
detail={"from": old, "to": u.project_role})
db.commit()
db.refresh(u)
return u.to_dict()
@app.delete("/api/auth/users/{user_id}")
def delete_user(user_id: str, admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
u = db.get(models.User, user_id)
@@ -545,7 +704,7 @@ def upsert_project(body: ProjectIn, user: models.User = Depends(auth.get_current
project_id=proj.id, summary=(proj.name or proj.number or proj.id))
db.commit()
# A project created by a non-admin auto-grants its creator access.
if is_new and user.role != "admin":
if is_new and not auth.is_admin(user):
grant_project_access(db, user.id, proj.id)
db.commit()
db.refresh(proj)
@@ -573,7 +732,10 @@ def delete_project(project_id: str, user: models.User = Depends(auth.get_current
proj = db.get(models.Project, project_id)
if not proj:
raise HTTPException(status_code=404, detail="Project not found")
require_project_access(db, user, proj.id)
# Cascades to every SOP and work package on the project — Project Admin only.
require_project_admin(db, user, proj.id, "Deleting a project")
log_event(db, user, "deleted", "project", proj.id, project_id=proj.id,
summary=(proj.name or proj.number or proj.id))
db.delete(proj)
db.commit()
return {"deleted": project_id}
@@ -587,6 +749,11 @@ def upsert_sop(body: SopIn, user: models.User = Depends(auth.get_current_user),
sop = db.get(models.Sop, body.id) if body.id else None
if sop is not None:
require_project_access(db, user, sop.project_id)
# The SOP is the project's baseline: once it's been completed, changing it
# is a Project Admin action. Authoring and revising a draft is open to any
# project member, including marking it complete the first time.
if sop.complete:
require_project_admin(db, user, sop.project_id, "Changing a completed SOP")
is_new = sop is None
if sop is None:
sop = models.Sop(id=body.id or gen_id("sop"))
@@ -645,7 +812,7 @@ def delete_sop(sop_id: str, user: models.User = Depends(auth.get_current_user),
sop = db.get(models.Sop, sop_id)
if not sop:
raise HTTPException(status_code=404, detail="SOP not found")
require_project_access(db, user, sop.project_id)
require_project_admin(db, user, sop.project_id, "Deleting a SOP")
log_event(db, user, "deleted", "sop", sop.id, project_id=sop.project_id,
summary=(sop.name or sop.number or sop.id))
db.delete(sop)
@@ -816,7 +983,9 @@ def delete_wp(wp_id: str, user: models.User = Depends(auth.get_current_user), db
wp = db.get(models.WorkPackage, wp_id)
if not wp:
raise HTTPException(status_code=404, detail="Work Package not found")
require_project_access(db, user, wp.project_id)
# Deleting a work package is irreversible — Project Admin only. A project_user
# who wants one out of the way can archive it instead (reversible).
require_project_admin(db, user, wp.project_id, "Deleting a work package")
log_event(db, user, "deleted", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id))
db.delete(wp)
@@ -924,6 +1093,13 @@ def get_app_settings(_admin: models.User = Depends(auth.require_admin), db: Sess
return notify.public_settings(db)
@app.get("/api/app-flags")
def get_app_flags(_user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""Feature flags every signed-in page reads (e.g. whether the BIM/VDC tooling
is switched on). No secrets — safe for any authenticated user."""
return notify.app_flags(db)
@app.put("/api/settings")
def put_app_settings(body: SettingsIn, admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
patch = {k: v for k, v in body.model_dump().items() if v is not None}
@@ -955,7 +1131,7 @@ def send_test_email(body: TestEmailIn, admin: models.User = Depends(auth.require
def list_notifications(all: bool = Query(False), limit: int = Query(100, ge=1, le=500),
user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
stmt = select(models.Notification)
if not (all and user.role == "admin"):
if not (all and auth.is_admin(user)):
stmt = stmt.where(models.Notification.user_id == user.id)
rows = db.scalars(stmt.order_by(models.Notification.created_at.desc()).limit(limit)).all()
return [n.to_dict() for n in rows]
@@ -967,13 +1143,15 @@ def project_members(project_id: str, user: models.User = Depends(auth.get_curren
require_project_access(db, user, project_id)
member_ids = set(db.scalars(select(models.ProjectMember.user_id).where(models.ProjectMember.project_id == project_id)).all())
members = db.scalars(select(models.User).where(models.User.id.in_(member_ids))).all() if member_ids else []
admins = db.scalars(select(models.User).where(models.User.role == "admin")).all()
admins = db.scalars(select(models.User).where(models.User.role == auth.ROLE_ADMIN)).all()
out, seen = [], set()
for u in list(members) + list(admins):
if u.id in seen or not u.is_active:
continue
seen.add(u.id)
out.append({"id": u.id, "username": u.username, "full_name": u.full_name, "email": u.email})
out.append({"id": u.id, "username": u.username, "full_name": u.full_name,
"email": u.email, "project_role": u.project_role or "",
"role": auth.normalize_role(u.role)})
out.sort(key=lambda x: (x["full_name"] or x["username"] or "").lower())
return out