fix: pre-car-test bug fixes from full code audit

- app_alerts.py: null guard on list_local_videos() result (.get('files', []))
- app_cellular.py: remove erroneous .encode(errors='replace') on str response body
- app_settings.py: initialize apn='' when isinstance check fails (NameError guard)
- app_upload.py: move json import to top of file instead of inside except handler
- server/app.py: wrap int(request.args.get('limit')) in try/except ValueError
- server/app.py: wrap os.path.commonpath() in try/except ValueError (cross-root paths)
- tests/test_app_settings.py: add missing trips_upload_enabled/video_convert_enabled
  to valid_payload() (fields added in Phase 23.5 but test not updated)

All 63 tests pass.

Co-authored-by: Copilot <[email protected]>
This commit is contained in:
Mathieu Moran
2026-08-10 05:15:27 -04:00
co-authored by Copilot
parent f30143c7b7
commit 1b39ac0cc8
6 changed files with 20 additions and 8 deletions
+1 -1
View File
@@ -231,7 +231,7 @@ def _power_condition(throttling, now=None):
def collect_conditions(config): def collect_conditions(config):
health = app_health.collect_health() health = app_health.collect_health()
videos = app_video.list_local_videos() videos = app_video.list_local_videos()
pending_videos = sum(1 for video in videos["files"] if not video["uploaded"]) pending_videos = sum(1 for video in (videos or {}).get("files", []) if not video["uploaded"])
pending_tracking = app_sql.count_unsent_tracking_rows() pending_tracking = app_sql.count_unsent_tracking_rows()
speed = app_obd.get_fresh_speed() speed = app_obd.get_fresh_speed()
fix_age = app_gps.get_fix_age() fix_age = app_gps.get_fix_age()
+1 -1
View File
@@ -229,7 +229,7 @@ def http_post(
break break
read_resp = _at(ser, "AT+HTTPREAD", terminators=("OK", "ERROR"), timeout=15) read_resp = _at(ser, "AT+HTTPREAD", terminators=("OK", "ERROR"), timeout=15)
body = read_resp.encode(errors="replace") body = read_resp
return status_code, body return status_code, body
finally: finally:
+1
View File
@@ -121,6 +121,7 @@ def update_settings(body):
apn = body.get("cellular_apn") apn = body.get("cellular_apn")
if not isinstance(apn, str): if not isinstance(apn, str):
errors["cellular_apn"] = "Must be text." errors["cellular_apn"] = "Must be text."
apn = ""
else: else:
apn = apn.strip() apn = apn.strip()
if len(apn) > 100 or not _APN_PATTERN.fullmatch(apn): if len(apn) > 100 or not _APN_PATTERN.fullmatch(apn):
+2 -3
View File
@@ -35,6 +35,7 @@ never commit real values - mirrors the db/creds.env pattern already used by
app_db.py for MongoDB). app_db.py for MongoDB).
""" """
import hashlib import hashlib
import json
import os import os
import time import time
from threading import Thread from threading import Thread
@@ -178,9 +179,7 @@ def _upload_tracking_batch():
if not app_cellular.is_configured(): if not app_cellular.is_configured():
raise raise
print(f"WiFi/Ethernet upload unreachable ({wifi_err}), trying cellular fallback...") print(f"WiFi/Ethernet upload unreachable ({wifi_err}), trying cellular fallback...")
import json as _json body_bytes = json.dumps({"points": points}).encode()
body_bytes = _json.dumps({"points": points}).encode()
try: try:
status_code, _body = app_cellular.http_post_json( status_code, _body = app_cellular.http_post_json(
f"{SERVER_URL}/api/v1/tracking/batch", body_bytes, _headers() f"{SERVER_URL}/api/v1/tracking/batch", body_bytes, _headers()
+11 -1
View File
@@ -430,7 +430,10 @@ def portal_trips():
device_id = _portal_device_id() device_id = _portal_device_id()
if device_id is None: if device_id is None:
return jsonify(error="device not configured"), 404 return jsonify(error="device not configured"), 404
try:
limit = min(int(request.args.get("limit", 200)), 1000) limit = min(int(request.args.get("limit", 200)), 1000)
except (ValueError, TypeError):
limit = 200
return jsonify(db.list_trips(device_id, limit=limit)) return jsonify(db.list_trips(device_id, limit=limit))
@@ -452,7 +455,10 @@ def portal_dashcam_list():
device_id = _portal_device_id() device_id = _portal_device_id()
if device_id is None: if device_id is None:
return jsonify(error="device not configured"), 404 return jsonify(error="device not configured"), 404
try:
limit = min(int(request.args.get("limit", 500)), 2000) limit = min(int(request.args.get("limit", 500)), 2000)
except (ValueError, TypeError):
limit = 500
return jsonify(db.list_dashcam_files(device_id, limit=limit)) return jsonify(db.list_dashcam_files(device_id, limit=limit))
@@ -468,7 +474,11 @@ def portal_dashcam_stream(file_id):
abort(404) abort(404)
storage_path = file_row["storage_path"] storage_path = file_row["storage_path"]
# Safety check: must still be under STORAGE_DIR # Safety check: must still be under STORAGE_DIR
if not os.path.commonpath([STORAGE_DIR, os.path.normpath(storage_path)]) == os.path.normpath(STORAGE_DIR): try:
safe = os.path.commonpath([STORAGE_DIR, os.path.normpath(storage_path)]) == os.path.normpath(STORAGE_DIR)
except ValueError:
safe = False
if not safe:
abort(403) abort(403)
if not os.path.isfile(storage_path): if not os.path.isfile(storage_path):
abort(404) abort(404)
+2
View File
@@ -8,6 +8,8 @@ from modules import app_cellular, app_settings
def valid_payload(): def valid_payload():
return { return {
"dashcam_upload_enabled": False, "dashcam_upload_enabled": False,
"trips_upload_enabled": True,
"video_convert_enabled": True,
"local_video_retention_days": 14, "local_video_retention_days": 14,
"cellular_enabled": True, "cellular_enabled": True,
"cellular_apn": "carrier.apn", "cellular_apn": "carrier.apn",