JWT, Rate Limiting & Request Hooks

JWT Configuration (app/__init__.py)

Token lifetimes and storage location are set once on app startup:

app.config["JWT_TOKEN_LOCATION"]       = ["headers", "query_string"]
app.config["JWT_SECRET_KEY"]           = Config.SECRET_KEY
app.config["JWT_ACCESS_TOKEN_EXPIRES"] = timedelta(hours=4)
app.config["JWT_REFRESH_TOKEN_EXPIRES"] = timedelta(days=14)

Setting	Current value	Notes
JWT_ACCESS_TOKEN_EXPIRES	4 hours	Change timedelta(hours=4) to adjust
JWT_REFRESH_TOKEN_EXPIRES	14 days	Default for web. Mobile tokens are issued with expires_delta=False — see note below
JWT_TOKEN_LOCATION	headers, query_string	Token accepted via Authorization: Bearer header or ?jwt= query param
JWT_SECRET_KEY	Azure Key Vault (SECRET_KEY)	Rotated in Key Vault; no code change needed

Mobile refresh token expiry

The global JWT_REFRESH_TOKEN_EXPIRES applies to web clients. For mobile logins, create_refresh_token() is called with expires_delta=False at the point of token issuance (inside the login service). To change mobile token lifetime, update that call — changing only app.config here will not affect mobile tokens.

Expired token response

warning

When an access token expires, Flask-JWT-Extended calls the registered @expired_token_loader, which returns status 420 (not the standard 401). Clients must detect 420 and use the refresh token to obtain a new access token.

@CachedData.jwt.expired_token_loader
def my_expired_token_callback(a, b):
    return jsonify({'status': 420, 'message': 'The token has expired'}), 420

user_lookup_loader — populating current_user

After the token passes signature and blocklist checks, @CachedData.jwt.user_lookup_loader runs to build the current_user object that every route accesses via flask_jwt_extended.current_user.

Industry resolution logic:

jwt_data['sub'] (industry_id)
    │
    ├── industry_id == INTERNAL_INDUSTRY_ID  AND  targetIndustryId header/param present?
    │       └── fetch industry by targetIndustryId   ← admin acting on a client industry
    │
    ├── industry_id == COMPANY_INDUSTRY_ID   AND  targetIndustryId header/param present?
    │       └── fetch industry by targetIndustryId   ← company-level user impersonating client
    │
    └── all other cases
            └── fetch industry by jwt_data['sub']    ← normal user, own industry

What gets built into current_user:

Field	Source
Industry fields (units, categories, etc.)	IndustryDataFormatter().formatIndustryData(industry_data, alerts_config)
userId	jwt_data['userId'] (from token claim)
permissions	jwt_data.get('permissions', []) (from token claim)
alertsConfig	DatabaseSupporter.get_alerts_config_by_industry_id()
leakageEnabled	True if any unit has alertEnabled.stable_flow == True in alerts config

note

Side effect: sets g.timezone = industry_data['meta']['timezone'] — this is what DateTimeUtil reads for all timezone-aware operations in that request.

Error path: If the industry document has deleted == True, the loader returns None. Flask-JWT-Extended then calls @user_lookup_error_loader, which aborts with 400 'Industry not found'.

---

Token blocklist

The @token_in_blocklist_loader is checked on every authenticated request (before user_lookup_loader). A token is blocked if:

• its userId claim is in Constants.disabled_user_ids
• its sub (industry ID) claim is in Constants.disabled_industry_ids

Both lists live in app/constants/constants.py. Add an entry to either list to immediately invalidate all tokens for that user or industry without a server restart — Flask reads from the in-memory Constants class on every request.

# app/constants/constants.py
disabled_user_ids = [
    'I2289_DEFAULT_sribrilluser',
]

disabled_industry_ids = [
    # add industryId here to block all tokens for that industry
]

Active token validation (TokenService)

Beyond the blocklist, non-INTERNAL tokens are validated against token_logs_container on every request (in validate_token_logic):

1. DatabaseSupporter.get_token_doc() fetches the token document by (industryId, userId, accessToken).
2. If not found, or isActive == False, or status != "ACTIVE" → returns 440 (session expired).
3. On success, updates lastActivity timestamp in the document.

info

A token is revoked at logout by setting isActive: false and status: "REVOKED" in the document. Admin (INTERNAL) tokens skip this check — they are validated by JWT signature only.

Token document ID format: {industryId}_{userId}_{deviceId}_{baseUrlHash}

---

Request / Response Hooks (app/__init__.py)

info

Every HTTP request passes through three @app.before_request hooks in registration order, then one @app.after_request hook on the way out.

Hook execution order

Incoming request
    │
    ├── 1. authenticate_and_authorize()   → auth_interceptor() — JWT check / public route bypass
    ├── 2. handle_time_zone()             → sets g.timezone fallback
    └── 3. track_user_ip()               → geo-IP logging for non-private IPs
    │
    [Route handler]
    │
    └── 4. add_security_headers()        → attaches security headers to every response

---

1. authenticate_and_authorize — JWT enforcement

@app.before_request
def authenticate_and_authorize():
    return auth_interceptor()

Delegates to auth_interceptor() in app/auth/auth.py. Full logic is covered in Public Route Handling.

---

2. handle_time_zone — timezone fallback

@app.before_request
def handle_time_zone():
    if not g.__contains__('timezone'):
        g.timezone = 'Asia/Kolkata'

note

g.timezone is normally set inside @jwt.user_lookup_loader from industry_data['meta']['timezone'] when a valid JWT is present. This hook runs as a safety net for public routes (where user_lookup_loader is never called) so that DateTimeUtil always has a timezone to read from g.

To change the default fallback timezone, update 'Asia/Kolkata' here.

---

3. track_user_ip — geo-IP logging (app/security_checks/track_user.py)

@app.before_request
def track_user_ip():
    track_ip()

track_ip() runs on every request and:

1. Resolves client IP from X-Forwarded-For (first value) or remote_addr.
2. Skips logging for private IPs (is_private_ip() check — RFC 1918 ranges).
3. For public IPs: calls http://ipinfo.io/{ip}/json (cached via @lru_cache(maxsize=1000)) to get city, region, country.
4. Writes a document to the log container via DatabaseSupporter.create_log_ip():

{
  "id": "2024-11-09T10:30:00Z_203.0.113.42",
  "userId": "IND001_DEFAULT_user",
  "industryId": "IND001",
  "created_on": "2024-11-09T10:30:00Z",
  "ip": "203.0.113.42",
  "city": "Mumbai",
  "region": "Maharashtra",
  "country": "IN"
}

note

userId and industryId are read from current_user — they will be "Unknown" on public routes where no JWT is present.

The @lru_cache on get_geo_info means each unique public IP is only looked up once per server lifetime (cache survives across requests but resets on restart).

---

4. add_security_headers — response headers

@app.after_request
def add_security_headers(response):
    response.headers['Content-Security-Policy'] = f"frame-ancestors {Config.ALLOWED_FRAME_ANCESTORS}"
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
    return response

Applied to every response regardless of route. The frame-ancestors value controls which origins may embed the app in an <iframe>. Current allowed origins are set in Config.ALLOWED_FRAME_ANCESTORS (read from the ALLOWED_FRAME_ANCESTORS environment variable, with a default covering all known Aquagen web apps):

'self'
https://web.aquagen.co.in
https://pre-prod-aquagen.web.app
https://test1-aquagen.web.app  https://test2-aquagen.web.app  https://test3-aquagen.web.app
https://aquagen-demo.web.app
https://smartcity-aquagen.web.app
http://localhost:4200

tip

To add a new origin, set or extend ALLOWED_FRAME_ANCESTORS in the environment — no code change needed.

---

Error Handlers

Handler	Trigger	Response
@app.errorhandler(ValidationError)	@validate_values() raises ValidationError on SQL injection pattern match	{"message": "..."} with status_code from the exception (default 405)
@app.errorhandler(429)	Flask-Limiter rate limit exceeded	{"message": "Too many requests...", "status": "failed"} with HTTP 429
@CachedData.jwt.expired_token_loader	JWT signature valid but token lifetime exceeded	{"status": 420, "message": "The token has expired"} with HTTP 420

---

Rate Limiting (app/util/rate_limiter.py)

info

Rate limiting uses Flask-Limiter with an in-memory fixed-window strategy.

Global defaults

Set on the Limiter instance:

limiter = Limiter(
    key_func=get_client_ip,
    storage_uri="memory://",
    default_limits=["5000 per day", "5000 per hour"],
    strategy="fixed-window"
)

note

These apply to every endpoint that does not have a more specific limit. To change the global cap, edit the default_limits list.

Key functions

Function	Keyed by	Used when
get_client_ip()	X-Forwarded-For header (first IP), or remote address	Default for all endpoints
get_phone_or_ip()	phNumber / username header if present, else IP	OTP endpoints (per-phone limiting)
get_ip_only()	IP only (ignores phone headers)	Login, user create

Per-endpoint limits (RATE_LIMITS)

Defined in app/util/rate_limiter.py. To change a limit, edit the string values here:

RATE_LIMITS = {
    "otp_generate": [
        "3 per minute",   # [0] applied with get_phone_or_ip key
        "10 per hour",    # [1] applied with get_phone_or_ip key
        "20 per day"      # [2] applied with get_phone_or_ip key
    ],
    "otp_verify": [
        "5 per minute",   # [0]
        "15 per hour"     # [1]
    ],
    "login": [
        "10 per minute",  # [0] applied with get_ip_only key
        "50 per hour"     # [1] applied with get_ip_only key
    ],
    "user_create": [
        "5 per hour",     # [0] applied with get_ip_only key
        "20 per day"      # [1] applied with get_ip_only key
    ]
}

Limits are applied in a route using the @limiter.limit() decorator:

from app.util.rate_limiter import limiter, get_ip_only, RATE_LIMITS

@userNamespace.route('/login')
class LoginRoute(Resource):
    @limiter.limit(RATE_LIMITS["login"][0], key_func=get_ip_only)
    @limiter.limit(RATE_LIMITS["login"][1], key_func=get_ip_only)
    def post(self):
        ...

note

Each @limiter.limit() call adds one independent window. All windows must pass — if any is exceeded, the request is rejected.

Limit string format

Flask-Limiter accepts: "N per period" where period is second, minute, hour, day, month, year.

Examples:

"10 per minute"
"100 per hour"
"1000 per day"
"5 per second"

Rate limit error response

When a limit is exceeded Flask returns 429. The error handler in app/__init__.py formats it:

@app.errorhandler(429)
def handle_rate_limit_error(error):
    return jsonify({
        'message': 'Too many requests. Please try again later.',
        'status': 'failed'
    }), 429

Adding a rate limit to a new endpoint

1. Add a key to RATE_LIMITS in app/util/rate_limiter.py:

   "my_feature": [
       "20 per minute",
       "200 per hour"
   ]

2. Import and apply in the route:

   from app.util.rate_limiter import limiter, get_ip_only, RATE_LIMITS
   
   @limiter.limit(RATE_LIMITS["my_feature"][0], key_func=get_ip_only)
   @limiter.limit(RATE_LIMITS["my_feature"][1], key_func=get_ip_only)
   def post(self):
       ...

Storage backend

warning

The current storage_uri="memory://" stores counters in-process. This means:

• Counters reset on server restart.
• Limits are not shared across multiple worker processes or instances.

To share limits across processes or pods (e.g. in a scaled deployment), switch to Redis:

limiter = Limiter(
    key_func=get_client_ip,
    storage_uri="redis://localhost:6379",
    default_limits=["5000 per day", "5000 per hour"],
    strategy="fixed-window"
)