Pre production

.github/workflows/deploy-to-gcp.yml

@@ -13,7 +13,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
 
       - name: Set environment
         run: |

.gitignore

@@ -229,3 +229,7 @@ cython_debug/
 # dependencies
 /node_modules
 /frontend/node_modules
+
+# Mise Envs
+mise.toml
+

README.md

@@ -20,6 +20,26 @@
 
 ---
 
+### Password Security
+
+New and changed passwords must be at least 12 characters and include lowercase
+letters, uppercase letters, numbers, and symbols. Passwords that include obvious
+account identifiers such as the username, email, full name, or display name are
+rejected.
+
+Existing weak passwords do not block users from signing in. After a successful
+sign-in, the application records the current password's last known strength
+status so the Settings page can warn the user if their current password is weak.
+The Settings password section also shows when the password was last changed.
+
+When users type a new password, the UI shows strength feedback immediately. On
+blur, the backend checks the candidate password against the Have I Been Pwned
+Pwned Passwords range API using k-anonymity: only the first five characters of a
+SHA-1 hash are sent, and the raw password is never sent to the third-party
+service. Known compromised passwords are rejected when a password is changed.
+
+---
+
 ### Tech Stack
 
 | Layer         | Technology           |

api/auth/dependencies.py

@@ -11,3 +11,6 @@ class ScopedUser(Enum):
     WellMeasurementWrite = scoped_user(["well_measurement:write"])
     MeterWrite = scoped_user(["meters:write"])
     WellWrite = scoped_user(["well:write"])
+    WorkOrderRead = scoped_user(["work_orders:read"])
+    WorkOrderCreate = scoped_user(["work_orders:create"])
+    WorkOrderUpdate = scoped_user(["work_orders:update"])

api/auth/password_policy.py

@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import hashlib
+import re
+import urllib.error
+import urllib.request
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from api.models.user import Users
+
+
+MIN_PASSWORD_LENGTH = 12
+HIBP_RANGE_API_URL = "https://api.pwnedpasswords.com/range"
+HIBP_USER_AGENT = "WaterManagerDB password checker"
+HIBP_TIMEOUT_SECONDS = 3
+
+
+@dataclass(frozen=True)
+class PasswordEvaluation:
+    score: int
+    label: str
+    is_policy_compliant: bool
+    missing_requirements: list[str]
+    compromised_count: int | None = None
+    compromised_checked_at: datetime | None = None
+    compromised_check_error: str | None = None
+
+
+def evaluate_password_strength(
+    password: str,
+    user: Users | None = None,
+    compromised_count: int | None = None,
+    compromised_checked_at: datetime | None = None,
+    compromised_check_error: str | None = None,
+) -> PasswordEvaluation:
+    missing_requirements = []
+
+    if len(password) < MIN_PASSWORD_LENGTH:
+        missing_requirements.append(
+            f"Use at least {MIN_PASSWORD_LENGTH} characters."
+        )
+    if not re.search(r"[a-z]", password):
+        missing_requirements.append("Add a lowercase letter.")
+    if not re.search(r"[A-Z]", password):
+        missing_requirements.append("Add an uppercase letter.")
+    if not re.search(r"\d", password):
+        missing_requirements.append("Add a number.")
+    if not re.search(r"[^A-Za-z0-9]", password):
+        missing_requirements.append("Add a symbol.")
+
+    lower_password = password.lower()
+    for value, label in _user_identifiers(user):
+        if value and len(value) >= 3 and value.lower() in lower_password:
+            missing_requirements.append(f"Do not include your {label}.")
+
+    score = 0
+    score += min(len(password), 16) // 4
+    score += 1 if re.search(r"[a-z]", password) else 0
+    score += 1 if re.search(r"[A-Z]", password) else 0
+    score += 1 if re.search(r"\d", password) else 0
+    score += 1 if re.search(r"[^A-Za-z0-9]", password) else 0
+    score += 1 if len(password) >= 16 else 0
+    score = min(score, 5)
+
+    if missing_requirements:
+        score = min(score, 2)
+
+    label = "Weak"
+    if score >= 5:
+        label = "Strong"
+    elif score >= 3:
+        label = "Moderate"
+
+    return PasswordEvaluation(
+        score=score,
+        label=label,
+        is_policy_compliant=not missing_requirements,
+        missing_requirements=missing_requirements,
+        compromised_count=compromised_count,
+        compromised_checked_at=compromised_checked_at,
+        compromised_check_error=compromised_check_error,
+    )
+
+
+def check_pwned_password(password: str) -> tuple[int | None, datetime, str | None]:
+    checked_at = datetime.now(timezone.utc)
+    password_hash = hashlib.sha1(password.encode("utf-8")).hexdigest().upper()
+    prefix = password_hash[:5]
+    suffix = password_hash[5:]
+    request = urllib.request.Request(
+        f"{HIBP_RANGE_API_URL}/{prefix}",
+        headers={
+            "Add-Padding": "true",
+            "User-Agent": HIBP_USER_AGENT,
+        },
+    )
+
+    try:
+        with urllib.request.urlopen(request, timeout=HIBP_TIMEOUT_SECONDS) as response:
+            body = response.read().decode("utf-8")
+    except (OSError, urllib.error.URLError, TimeoutError) as exc:
+        return None, checked_at, str(exc)
+
+    for line in body.splitlines():
+        found_suffix, _, count = line.partition(":")
+        if found_suffix == suffix:
+            try:
+                return int(count), checked_at, None
+            except ValueError:
+                return None, checked_at, "Unexpected compromised password response."
+
+    return 0, checked_at, None
+
+
+def evaluate_password(
+    password: str,
+    user: Users | None = None,
+    include_compromised_check: bool = False,
+) -> PasswordEvaluation:
+    compromised_count = None
+    compromised_checked_at = None
+    compromised_check_error = None
+
+    if include_compromised_check:
+        compromised_count, compromised_checked_at, compromised_check_error = (
+            check_pwned_password(password)
+        )
+
+    return evaluate_password_strength(
+        password=password,
+        user=user,
+        compromised_count=compromised_count,
+        compromised_checked_at=compromised_checked_at,
+        compromised_check_error=compromised_check_error,
+    )
+
+
+def apply_password_evaluation(user: Users, evaluation: PasswordEvaluation) -> None:
+    user.password_strength_score = evaluation.score
+    user.password_strength_label = evaluation.label
+    user.password_policy_compliant = evaluation.is_policy_compliant
+
+    if evaluation.compromised_checked_at is not None:
+        user.password_compromised_checked_at = evaluation.compromised_checked_at
+        user.password_compromised_count = evaluation.compromised_count
+
+
+def _user_identifiers(user: Users | None) -> list[tuple[str | None, str]]:
+    if user is None:
+        return []
+
+    values: list[tuple[str | None, str]] = [
+        (user.username, "username"),
+        (user.email, "email"),
+        (user.full_name, "name"),
+        (user.display_name, "display name"),
+    ]
+
+    if user.email and "@" in user.email:
+        values.append((user.email.split("@", 1)[0], "email"))
+
+    return values

api/auth/session_tracking.py

@@ -5,8 +5,10 @@
 from uuid import uuid4
 
 from fastapi import Request
+from sqlalchemy import or_
 from sqlalchemy.orm import Session
 
+from api.config import settings
 from api.models.user import SignOutReasonTypeLU, UserSessions, Users
 
 LAST_SEEN_UPDATE_INTERVAL = timedelta(minutes=5)
@@ -105,6 +107,87 @@ def build_device_label(
     return browser or operating_system or device_type
 
 
+def close_user_session(
+    db: Session,
+    session: UserSessions,
+    reason_name: Optional[str],
+    signed_out_at: datetime,
+) -> UserSessions:
+    if session.signed_out_at is not None:
+        return session
+
+    sign_out_reason = get_sign_out_reason(db, reason_name)
+
+    session.signed_out_at = signed_out_at
+    session.last_seen_at = signed_out_at
+    session.is_active = False
+    session.sign_out_reason_type_id = sign_out_reason.id if sign_out_reason else None
+    db.add(session)
+
+    return session
+
+
+def close_old_active_user_sessions(
+    db: Session, user: Users, signed_out_at: datetime
+) -> None:
+    retention_days = max(settings.USER_SESSION_RETENTION_DAYS, 0)
+    cutoff = signed_out_at - timedelta(days=retention_days)
+    old_sessions = (
+        db.query(UserSessions)
+        .filter(
+            UserSessions.user_id == user.id,
+            UserSessions.is_active.is_(True),
+            UserSessions.signed_out_at.is_(None),
+            or_(
+                UserSessions.signed_in_at < cutoff,
+                UserSessions.last_seen_at < cutoff,
+            ),
+        )
+        .all()
+    )
+
+    for session in old_sessions:
+        close_user_session(db, session, "session_expired", signed_out_at)
+
+
+def close_existing_machine_session(
+    db: Session,
+    user: Users,
+    signed_out_at: datetime,
+    user_agent: Optional[str],
+    device_label: Optional[str],
+    device_type: Optional[str],
+    browser: Optional[str],
+    operating_system: Optional[str],
+    fingerprint_hash: Optional[str],
+) -> None:
+    query = db.query(UserSessions).filter(
+        UserSessions.user_id == user.id,
+        UserSessions.is_active.is_(True),
+        UserSessions.signed_out_at.is_(None),
+    )
+
+    if fingerprint_hash:
+        matching_sessions = query.filter(
+            UserSessions.fingerprint_hash == fingerprint_hash
+        ).all()
+    else:
+        if not any([user_agent, device_label, device_type, browser, operating_system]):
+            return
+
+        matching_sessions = query.filter(
+            UserSessions.fingerprint_hash.is_(None),
+            UserSessions.user_agent == user_agent,
+            UserSessions.device_label == device_label,
+            UserSessions.device_type == device_type,
+            UserSessions.browser == browser,
+            UserSessions.operating_system == operating_system,
+        ).all()
+
+    for session in matching_sessions:
+        close_user_session(db, session, "forced_logout", signed_out_at)
+
+
 def create_user_session(db: Session, user: Users, request: Request) -> UserSessions:
     user_agent = normalize_header_value(request.headers.get("user-agent"))
     browser = normalize_header_value(request.headers.get("x-browser")) or parse_browser(
@@ -122,6 +205,20 @@ def create_user_session(db: Session, user: Users, request: Request) -> UserSessi
     fingerprint_hash = normalize_header_value(
         request.headers.get("x-device-fingerprint")
     )
+    now = datetime.utcnow()
+
+    close_old_active_user_sessions(db=db, user=user, signed_out_at=now)
+    close_existing_machine_session(
+        db=db,
+        user=user,
+        signed_out_at=now,
+        user_agent=user_agent,
+        device_label=device_label,
+        device_type=device_type,
+        browser=browser,
+        operating_system=operating_system,
+        fingerprint_hash=fingerprint_hash,
+    )
 
     session = UserSessions(
         user_id=user.id,
@@ -133,8 +230,8 @@ def create_user_session(db: Session, user: Users, request: Request) -> UserSessi
         browser=browser,
         operating_system=operating_system,
         fingerprint_hash=fingerprint_hash,
-        signed_in_at=datetime.utcnow(),
-        last_seen_at=datetime.utcnow(),
+        signed_in_at=now,
+        last_seen_at=now,
         is_active=True,
     )
 
@@ -188,15 +285,7 @@ def mark_session_signed_out(
     if session.signed_out_at is not None:
         return session
 
-    sign_out_reason = get_sign_out_reason(db, reason_name)
-
-    session.signed_out_at = datetime.utcnow()
-    session.last_seen_at = session.signed_out_at
-    session.is_active = False
-    session.sign_out_reason_type_id = sign_out_reason.id if sign_out_reason else None
-    db.add(session)
-
-    return session
+    return close_user_session(db, session, reason_name, datetime.utcnow())
 
 
 def touch_user_session(db: Session, session_identifier: Optional[str]) -> None:

api/config.py

@@ -30,6 +30,9 @@ class Settings:
     JWT_SECRET_KEY: Optional[str] = os.getenv("JWT_SECRET_KEY")
     JWT_ALGORITHM: str = os.getenv("JWT_ALGORITHM", "HS256")
     ACCESS_TOKEN_EXPIRE_HOURS: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_HOURS", "8"))
+    USER_SESSION_RETENTION_DAYS: int = int(
+        os.getenv("USER_SESSION_RETENTION_DAYS", "14")
+    )
     APP_ENV: str = os.getenv(
         "APP_ENV",
         "production" if os.getenv("APPDB_ENV") == ".env_production" else "development",