04/27 Fixed some issues 2

app/__init__.py

@@ -19,7 +19,10 @@ csrf          = CSRFProtect()    # initialized here; .init_app() called in creat
 limiter       = Limiter(
     key_func       = get_remote_address,
     default_limits = [],          # no global limit — applied per-route only
-    storage_uri    = 'memory://', # in-process; swap for 'redis://...' in multi-worker setups
+    # Use Redis when REDIS_URL is set in the environment (production multi-worker).
+    # Falls back to in-process memory for local development (single-worker only;
+    # counters are NOT shared across Gunicorn workers in memory:// mode).
+    storage_uri    = os.environ.get('REDIS_URL', 'memory://'),
 )
 
 

app/models/user.py

@@ -69,9 +69,17 @@ class User(UserMixin, db.Model):
 
     @staticmethod
     def verify_set_password_token(token):
-        """Return the User whose token matches, or None if invalid/expired."""
+        """Return the User whose token matches, or None if invalid/expired.
+
+        The final token comparison uses hmac.compare_digest so that the
+        comparison runs in constant time regardless of how many characters
+        match, preventing timing-based token enumeration attacks.
+        """
+        import hmac
         if not token:
             return None
+        # Primary lookup is via DB index — compare_digest is a defense-in-depth
+        # guard applied after the row is retrieved to harden the string comparison.
         user = User.query.filter_by(set_password_token=token).first()
         if user is None:
             return None
@@ -79,7 +87,10 @@ class User(UserMixin, db.Model):
             return None
         if now_eastern() > user.set_password_token_expires:
             return None
+        # Constant-time comparison — prevents timing oracle on the stored token
+        if not hmac.compare_digest(user.set_password_token, token):
+            return None
         return user
 
     def __repr__(self):
-        return f'<User {self.username}>'
+        return f'<User {self.username}>'

app/utils/scope.py

@@ -48,25 +48,32 @@ def get_customer_scope(user) -> list[int] | None:
 
     assignments = CustomerAssignment.query.filter_by(user_id=user.id).all()
 
-    facility_ids = set()
+    if not assignments:
+        return []
 
-    for assignment in assignments:
+    # Separate direct facility assignments from project-level assignments
-        if assignment.facility_id:
+    direct_facility_ids = {a.facility_id for a in assignments if a.facility_id}
-            # Scoped to a specific facility
+    project_ids         = {a.project_id  for a in assignments if not a.facility_id}
-            facility_ids.add(assignment.facility_id)
+
-        else:
+    facility_ids = set(direct_facility_ids)
-            # Scoped to an entire project — include all facilities in that project
+
-            project_facilities = (
+    # Single bulk query for all project-scoped facilities — replaces the
-                Facility.query
+    # previous per-assignment Facility.query loop (N+1 pattern).
-                .filter_by(project_id=assignment.project_id, active=True)
+    if project_ids:
-                .all()
+        project_facilities = (
+            Facility.query
+            .filter(
+                Facility.project_id.in_(project_ids),
+                Facility.active == True,
             )
-            for f in project_facilities:
+            .all()
-                facility_ids.add(f.id)
+        )
+        for f in project_facilities:
+            facility_ids.add(f.id)
 
     logger.debug(
         'SCOPE | customer_scope | user_id=%s username=%s facility_ids=%s',
         user.id, user.username, sorted(facility_ids),
     )
 
-    return sorted(facility_ids)
+    return sorted(facility_ids)

requirements.txt

@@ -5,6 +5,7 @@ Flask-WTF
 Flask-Mail
 Flask-Migrate
 Flask-Limiter
+redis
 PyMySQL
 cryptography
 python-dotenv