nngo/LT_Janitorial_Quality_Control
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

March 17 2026: Mobile app phase 1

Browse Source
This commit is contained in:
nngo
2026-03-17 17:24:55 -04:00
parent 28196b4a03
commit 75a7b26695
10 changed files with 873 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/__init__.py
+60
View File
@@ -0,0 +1,60 @@
"""
app/api/__init__.py
-------------------
Registers the /api/v1 blueprint group.
All mobile API routes live under the prefix /api/v1/.
This module is imported once from app/__init__.py — see the integration
instructions at the bottom of this file.
Blueprint layout
----------------
/api/v1/auth/login → api_auth.login
/api/v1/auth/refresh → api_auth.refresh
/api/v1/auth/logout → api_auth.logout
/api/v1/auth/me → api_auth.me
/api/v1/devices/register → api_auth.register_device
Future phases will add:
/api/v1/facilities → api_facilities.*
/api/v1/inspections → api_inspections.*
/api/v1/issues → api_issues.*
/api/v1/notifications → api_notifications.*
/api/v1/photos/upload → api_photos.*
"""
from flask import Blueprint
from app.api.errors import register_error_handlers
# Parent blueprint — all sub-blueprints registered under this prefix
api_bp = Blueprint('api', __name__, url_prefix='/api/v1')
# Register JSON error handlers so Flask exceptions within /api/v1/
# return JSON instead of HTML error pages.
register_error_handlers(api_bp)
def register_api(app):
"""
Import and register all API sub-blueprints onto api_bp, then
register api_bp on the Flask app.
Called once from create_app() in app/__init__.py.
"""
# ── Phase 1: Auth ────────────────────────────────────────────────────
from app.api.auth import bp as auth_bp
api_bp.register_blueprint(auth_bp)
# ── Phase 2+: Additional blueprints registered here as phases complete
# from app.api.facilities import bp as facilities_bp
# from app.api.inspections import bp as inspections_bp
# from app.api.issues import bp as issues_bp
# from app.api.notifications import bp as notifications_bp
# from app.api.photos import bp as photos_bp
# api_bp.register_blueprint(facilities_bp)
# api_bp.register_blueprint(inspections_bp)
# api_bp.register_blueprint(issues_bp)
# api_bp.register_blueprint(notifications_bp)
# api_bp.register_blueprint(photos_bp)
app.register_blueprint(api_bp)
app/api/auth.py
+319
View File
@@ -0,0 +1,319 @@
"""
app/api/auth.py
---------------
Authentication endpoints for the JQC mobile app.
POST /api/v1/auth/login
Accepts username + password.
Returns a short-lived access token (JWT) and a long-lived refresh token
(opaque, stored in DB). The app stores both in the iOS Keychain.
POST /api/v1/auth/refresh
Accepts a refresh token.
Returns a new access token. The refresh token is rotated — the old one
is revoked and a new one is issued, preventing replay attacks.
POST /api/v1/auth/logout
Accepts a refresh token.
Revokes it so it can no longer be used to issue new access tokens.
The app should discard both tokens from the Keychain after this call.
POST /api/v1/devices/register
Registers or updates the APNs device token for push notifications.
Called on every app launch after the user has already authenticated.
Requires a valid access token (JWT).
GET /api/v1/auth/me
Returns the current user's profile from the access token.
Useful for the app to verify the token is still valid on launch.
"""
import logging
from flask import Blueprint, request, g
from flask_wtf.csrf import csrf_exempt
from app import db
from app.models.user import User
from app.models.api_token import RefreshToken, DeviceToken
from app.api.errors import api_ok, api_error
from app.api.jwt_utils import generate_access_token
from app.api.decorators import jwt_required
from app.utils.audit import log_action, ACTION_LOGIN, ACTION_LOGOUT
from app.utils.time_utils import now_eastern
logger = logging.getLogger(__name__)
bp = Blueprint('api_auth', __name__)
def _user_payload(user: User) -> dict:
"""Serialize a User to the dict returned in auth responses."""
return {
'id': user.id,
'username': user.username,
'email': user.email,
'role': user.role,
'created_at': user.created_at.isoformat() if user.created_at else None,
}
# ── Login ─────────────────────────────────────────────────────────────────────
@bp.route('/auth/login', methods=['POST'])
@csrf_exempt
def login():
"""
Authenticate with username + password.
Request JSON
------------
{
"username": "john",
"password": "secret",
"device_id": "A1B2C3D4...", // UIDevice.identifierForVendor (optional)
"device_name": "John's iPhone" // (optional)
}
Response 200
------------
{
"ok": true,
"data": {
"access_token": "<jwt>",
"refresh_token": "<opaque_hex>",
"token_type": "Bearer",
"expires_in": 3600,
"user": { id, username, email, role, created_at }
}
}
"""
data = request.get_json(silent=True) or {}
username = (data.get('username') or '').strip()
password = data.get('password') or ''
if not username or not password:
return api_error('username and password are required', 400)
user = User.query.filter_by(username=username).first()
# Generic message — never reveal whether the username exists
if user is None or not user.check_password(password):
logger.warning('API login failed | username=%s | ip=%s',
username, request.remote_addr)
return api_error('Invalid credentials', 401)
if not user.active:
return api_error('Account is disabled. Please contact an administrator.', 401)
device_id = (data.get('device_id') or '')[:64] or None
device_name = (data.get('device_name') or '')[:100] or None
# Issue tokens
access_token = generate_access_token(user)
raw_refresh, rt_row = RefreshToken.create_for(
user,
device_id=device_id,
device_name=device_name,
)
db.session.commit()
log_action(ACTION_LOGIN, 'User', user.id, user.username,
f'source=mobile_api; device_id={device_id}')
logger.info('API LOGIN | user=%s | role=%s | device_id=%s',
user.username, user.role, device_id)
return api_ok({
'access_token': access_token,
'refresh_token': raw_refresh,
'token_type': 'Bearer',
'expires_in': 3600, # seconds — matches ACCESS_TOKEN_LIFETIME_MINUTES * 60
'user': _user_payload(user),
})
# ── Refresh ───────────────────────────────────────────────────────────────────
@bp.route('/auth/refresh', methods=['POST'])
@csrf_exempt
def refresh():
"""
Exchange a valid refresh token for a new access token.
The refresh token is rotated on every call — the submitted token is
revoked and a brand new one is issued. This limits the damage window
if a token is ever stolen.
Request JSON
------------
{ "refresh_token": "<opaque_hex>" }
Response 200
------------
{
"ok": true,
"data": {
"access_token": "<new_jwt>",
"refresh_token": "<new_opaque_hex>",
"token_type": "Bearer",
"expires_in": 3600
}
}
"""
data = request.get_json(silent=True) or {}
raw_token = (data.get('refresh_token') or '').strip()
if not raw_token:
return api_error('refresh_token is required', 400)
rt_row = RefreshToken.verify(raw_token)
if rt_row is None:
logger.warning('API refresh rejected | invalid/expired token | ip=%s',
request.remote_addr)
return api_error('Refresh token is invalid or expired', 401)
user = User.query.get(rt_row.user_id)
if user is None or not user.active:
rt_row.revoke()
db.session.commit()
return api_error('Account not available', 401)
# Rotate: revoke old token, issue new pair
device_id = rt_row.device_id
device_name = rt_row.device_name
rt_row.revoke()
new_access = generate_access_token(user)
new_raw_refresh, new_rt = RefreshToken.create_for(
user,
device_id=device_id,
device_name=device_name,
)
db.session.commit()
logger.info('API TOKEN REFRESH | user=%s | device_id=%s',
user.username, device_id)
return api_ok({
'access_token': new_access,
'refresh_token': new_raw_refresh,
'token_type': 'Bearer',
'expires_in': 3600,
})
# ── Logout ────────────────────────────────────────────────────────────────────
@bp.route('/auth/logout', methods=['POST'])
@csrf_exempt
@jwt_required
def logout():
"""
Revoke the current session's refresh token.
The app should call this when the user taps "Log out" and then discard
both the access token and refresh token from the Keychain.
Request JSON
------------
{ "refresh_token": "<opaque_hex>" }
Response 200
------------
{ "ok": true, "data": { "message": "Logged out" } }
"""
data = request.get_json(silent=True) or {}
raw_token = (data.get('refresh_token') or '').strip()
if raw_token:
rt_row = RefreshToken.verify(raw_token)
if rt_row and rt_row.user_id == g.api_user.id:
rt_row.revoke()
db.session.commit()
log_action(ACTION_LOGOUT, 'User', g.api_user.id, g.api_user.username,
'source=mobile_api')
logger.info('API LOGOUT | user=%s', g.api_user.username)
return api_ok({'message': 'Logged out successfully'})
# ── Current user ──────────────────────────────────────────────────────────────
@bp.route('/auth/me', methods=['GET'])
@jwt_required
def me():
"""
Return the authenticated user's profile.
Called by the app on launch to verify the stored access token is still
valid and to refresh the local user record.
Response 200
------------
{ "ok": true, "data": { "user": { id, username, email, role, ... } } }
"""
return api_ok({'user': _user_payload(g.api_user)})
# ── Device token registration ─────────────────────────────────────────────────
@bp.route('/devices/register', methods=['POST'])
@csrf_exempt
@jwt_required
def register_device():
"""
Register or update the APNs device token for the authenticated user.
Called on every app launch after authentication so the server always
has the current token (APNs rotates tokens periodically).
Request JSON
------------
{
"device_id": "<UIDevice.identifierForVendor>",
"apns_token": "<hex_string_from_didRegisterForRemoteNotifications>",
"device_name": "John's iPhone", // optional
"app_version": "1.0.3" // optional
}
Response 200
------------
{ "ok": true, "data": { "registered": true } }
"""
data = request.get_json(silent=True) or {}
device_id = (data.get('device_id') or '').strip()[:64]
apns_token = (data.get('apns_token') or '').strip()[:200]
device_name = (data.get('device_name') or '').strip()[:100] or None
app_version = (data.get('app_version') or '').strip()[:20] or None
if not device_id or not apns_token:
return api_error('device_id and apns_token are required', 400)
# Upsert: update existing row or insert new one
existing = DeviceToken.query.filter_by(
user_id=g.api_user.id,
device_id=device_id,
).first()
if existing:
existing.apns_token = apns_token
existing.device_name = device_name
existing.app_version = app_version
existing.registered_at = now_eastern()
else:
db.session.add(DeviceToken(
user_id = g.api_user.id,
device_id = device_id,
apns_token = apns_token,
device_name = device_name,
app_version = app_version,
))
db.session.commit()
logger.info('API DEVICE REGISTERED | user=%s | device_id=%s | apns_token=...%s',
g.api_user.username, device_id, apns_token[-6:])
return api_ok({'registered': True})
app/api/decorators.py
+108
View File
@@ -0,0 +1,108 @@
"""
app/api/decorators.py
---------------------
Request-level guards for all /api/v1/ endpoints.
@jwt_required
Validates the Bearer token in the Authorization header.
On success, sets flask.g.api_user to the authenticated User instance
so any route can access it without a second DB query.
@api_role_required(*roles)
Must be applied AFTER @jwt_required.
Rejects callers whose role is not in the allowed set.
Usage
-----
@bp.route('/inspections')
@jwt_required
def list_inspections():
user = g.api_user
...
@bp.route('/admin/users')
@jwt_required
@api_role_required('admin')
def admin_only():
...
"""
import logging
from functools import wraps
from flask import request, g, abort
from app.api.jwt_utils import decode_access_token
from app.api.errors import api_error
from app.models.user import User
logger = logging.getLogger(__name__)
def jwt_required(f):
"""
Validate the JWT Bearer token and load the user into flask.g.api_user.
Returns 401 if:
- Authorization header is missing or malformed
- Token is expired or invalid
- User referenced by the token no longer exists
- User account has been disabled (active=False)
"""
@wraps(f)
def decorated(*args, **kwargs):
auth_header = request.headers.get('Authorization', '')
if not auth_header.startswith('Bearer '):
return api_error('Missing or malformed Authorization header', 401)
raw_token = auth_header[len('Bearer '):]
payload = decode_access_token(raw_token)
if payload is None:
return api_error('Access token is invalid or expired', 401)
user_id = int(payload.get('sub', 0))
user = User.query.get(user_id)
if user is None:
return api_error('User not found', 401)
if not user.active:
return api_error('Account is disabled', 401)
# Make the user available to the route without re-querying
g.api_user = user
return f(*args, **kwargs)
return decorated
def api_role_required(*roles):
"""
Restrict an endpoint to users whose role is in the provided list.
Must be stacked BELOW @jwt_required so that g.api_user is already set.
Example
-------
@jwt_required
@api_role_required('admin', 'supervisor')
def supervisor_only_route():
...
"""
def decorator(f):
@wraps(f)
def decorated(*args, **kwargs):
user = getattr(g, 'api_user', None)
if user is None:
# Defensive: jwt_required should always run first
return api_error('Authentication required', 401)
if user.role not in roles:
logger.warning(
'API role denied | user=%s role=%s required=%s endpoint=%s',
user.username, user.role, roles, request.endpoint,
)
return api_error('Insufficient permissions', 403)
return f(*args, **kwargs)
return decorated
return decorator
app/api/errors.py
+87
View File
@@ -0,0 +1,87 @@
"""
app/api/errors.py
-----------------
Consistent JSON error responses for every API endpoint.
Every response — success or failure — uses the same envelope:
{
"ok": true | false,
"data": { ... } | null,
"error": null | "Human-readable message"
}
Usage
-----
from app.api.errors import api_error, api_ok
return api_ok({'inspection': {...}})
return api_error('Inspection not found', 404)
"""
from flask import jsonify
def api_ok(data=None, status=200):
"""Return a successful JSON response."""
return jsonify({
'ok': True,
'data': data,
'error': None,
}), status
def api_error(message: str, status: int = 400):
"""Return an error JSON response."""
return jsonify({
'ok': False,
'data': None,
'error': message,
}), status
# ── Registered error handlers (attached to the api blueprint) ─────────────────
def register_error_handlers(bp):
"""
Attach JSON error handlers to the given blueprint so that Flask
exceptions (404, 405, 500, etc.) return JSON instead of HTML within
the /api/v1/ prefix.
"""
from werkzeug.exceptions import HTTPException
@bp.errorhandler(400)
def bad_request(e):
return api_error(str(e.description) if hasattr(e, 'description') else 'Bad request', 400)
@bp.errorhandler(401)
def unauthorized(e):
return api_error('Authentication required', 401)
@bp.errorhandler(403)
def forbidden(e):
return api_error('Access denied', 403)
@bp.errorhandler(404)
def not_found(e):
return api_error('Resource not found', 404)
@bp.errorhandler(405)
def method_not_allowed(e):
return api_error('Method not allowed', 405)
@bp.errorhandler(413)
def payload_too_large(e):
return api_error('Uploaded file is too large', 413)
@bp.errorhandler(422)
def unprocessable(e):
return api_error('Unprocessable request', 422)
@bp.errorhandler(500)
def internal_error(e):
return api_error('Internal server error', 500)
@bp.errorhandler(HTTPException)
def generic_http(e):
return api_error(e.description or e.name, e.code)
app/api/jwt_utils.py
+74
View File
@@ -0,0 +1,74 @@
"""
app/api/jwt_utils.py
--------------------
Thin wrappers around PyJWT for signing and verifying access tokens.
Access tokens are short-lived JWTs (default 60 minutes) signed with
HMAC-SHA256 using the app's SECRET_KEY. They carry only the minimum
claims needed to identify the caller:
{
"sub": "42", # user.id as string
"role": "inspector", # user.role
"iat": 1710000000, # issued-at (UTC epoch)
"exp": 1710003600, # expiry (UTC epoch, 60 min later)
}
Refresh tokens are opaque random strings stored in the DB
(see app/models/api_token.py). This module only handles JWTs.
"""
import logging
from datetime import datetime, timezone, timedelta
import jwt
from flask import current_app
logger = logging.getLogger(__name__)
ACCESS_TOKEN_LIFETIME_MINUTES = 60
def _secret():
return current_app.config['SECRET_KEY']
def generate_access_token(user, lifetime_minutes: int = ACCESS_TOKEN_LIFETIME_MINUTES) -> str:
"""
Create and sign a new access token for the given user.
Parameters
----------
user : User ORM instance
lifetime_minutes : Token validity window (default 60 min)
Returns
-------
str
Signed JWT string ready to include in Authorization header.
"""
now = datetime.now(timezone.utc)
payload = {
'sub': str(user.id),
'role': user.role,
'iat': now,
'exp': now + timedelta(minutes=lifetime_minutes),
}
return jwt.encode(payload, _secret(), algorithm='HS256')
def decode_access_token(token: str) -> dict | None:
"""
Decode and verify a JWT access token.
Returns the payload dict on success, or None if the token is invalid,
expired, or tampered with. Logs the failure reason at DEBUG level.
"""
try:
return jwt.decode(token, _secret(), algorithms=['HS256'])
except jwt.ExpiredSignatureError:
logger.debug('JWT decode failed: token expired')
return None
except jwt.InvalidTokenError as exc:
logger.debug('JWT decode failed: %s', exc)
return None