feat(gsm): restrict unauthorized inbound SMS sessions

GSM-module/GSM-fastapi/.env.example

@@ -11,6 +11,21 @@ LOG_LEVEL=INFO
 # Maximum outbound SMS requests waiting behind the one in-flight request.
 # Must be an integer from 1 through 20.
 SMS_SEND_QUEUE_MAXSIZE=10
+# Maximum inbound events waiting for serial relay handling. Excess events are
+# dropped and counted in /health/detailed rather than consuming memory.
+SMS_INCOMING_QUEUE_MAXSIZE=100
+# Hard carrier-spend ceiling. Set to 0 to disable all outgoing SMS immediately.
+SMS_DAILY_SEND_LIMIT=100
+# Per-sender daily reply ceiling and sender-target relay ceiling.
+SMS_SENDER_DAILY_LIMIT=20
+SMS_SENDER_TARGET_DAILY_LIMIT=10
+# Minimum spacing between replies in the same response category for one sender.
+SMS_RESPONSE_COOLDOWN_SECONDS=30
+# Retain redacted operational SMS logs for this many days. Set to 0 to purge
+# existing logs at each service start.
+SMS_LOG_RETENTION_DAYS=30
+LOG_MAX_BYTES=1000000
+LOG_BACKUP_COUNT=3
 SAPOT_API_URL=http://localhost:8000
 GSM_SECRET=change-me-to-a-strong-secret
 SMS_BOT_USER_ID=

GSM-module/GSM-fastapi/api.py

@@ -10,20 +10,20 @@
   - Shuts everything down cleanly on exit
 
 Endpoints
-  GET  /health                 — liveness + GSM status
-  GET  /health/detailed        — full diagnostic info
+  GET  /health                 -- liveness + GSM status
+  GET  /health/detailed        -- full diagnostic info
 
-  POST /sms/send               — send an SMS (blocks until delivered)
-  GET  /sms/messages           — message log
+  POST /sms/send               -- send an SMS (blocks until delivered)
+  GET  /sms/messages           -- message log
 
-  GET  /users                  — list registered SAPOT users
-  POST /users                  — add a user
-  GET  /users/{phone}          — look up a user
+  GET  /users                  -- list registered SAPOT users
+  POST /users                  -- add a user
+  GET  /users/{phone}          -- look up a user
 
-  GET  /sessions               — all active sessions (debug)
-  DELETE /sessions/{phone}     — reset a session
+  GET  /sessions               -- all active sessions (debug)
+  DELETE /sessions/{phone}     -- reset a session
 
-  GET  /status                 — modem + serial status
+  GET  /status                 -- modem + serial status
 """
 
 import asyncio
@@ -68,12 +68,16 @@ async def lifespan(app: FastAPI):
             "Marked %d orphaned SMS messages as failed after service restart",
             orphaned_count,
         )
+    purged_count = database.purge_expired_sms_logs(settings.sms_log_retention_days)
+    if purged_count:
+        logger.info("Purged %d expired SMS log records", purged_count)
 
     # Serial worker
     _worker = SerialWorker(
         settings.serial_port,
         settings.serial_baud,
         settings.sms_send_queue_maxsize,
+        settings.sms_incoming_queue_maxsize,
     )
     _worker.start()
     logger.info("Serial worker started on %s", settings.serial_port)
@@ -141,26 +145,42 @@ def _process_incoming(event):
     if rejection_reason:
         database.update_message_status(msg_id, "rejected", rejection_reason)
 
-    # Forward to target via SMS
+    if reply and not database.allow_inbound_response(
+        number,
+        rejection_reason or "REPLY",
+        settings.sms_sender_daily_limit,
+        settings.sms_response_cooldown_seconds,
+    ):
+        database.update_message_status(msg_id, "rejected", "RATE_LIMITED")
+        return
+
+    # Forward to target via SMS. A confirmation is only sent if the target SMS
+    # was accepted by the modem, so users do not receive a false "Forwarded" state.
+    target_sent = True
     if forward_number and forward_body and _worker:
-        _send_and_log(
+        target_sent = _send_and_log(
             from_number="SERVER",
             to_number=forward_number,
             body=forward_body,
         )
 
+    if not target_sent:
+        reply = "Could not forward your message. Please try again."
+
 
     # Send reply back to sender
     if reply and _worker:
-        _send_and_log(
+        reply_sent = _send_and_log(
             from_number="SERVER",
             to_number=number,
             body=reply,
         )
+        if reply_sent and rejection_reason == "NO_ACCOUNT":
+            database.mark_unregistered_warning(number)
 
 
 
-def _send_and_log(from_number: str, to_number: str, body: str):
+def _send_and_log(from_number: str, to_number: str, body: str) -> bool:
     """Send one SMS via the worker and persist result to DB."""
     msg_id = database.log_message(
         direction="OUT",
@@ -169,19 +189,32 @@ def _send_and_log(from_number: str, to_number: str, body: str):
         body=body,
         status="pending",
     )
+    if database.is_sms_opted_out(to_number):
+        database.update_message_status(msg_id, "failed", "RECIPIENT_OPTED_OUT")
+        return False
+    if not database.reserve_outbound_sms(settings.sms_daily_send_limit):
+        logger.warning("Outbound SMS rejected: daily send limit reached")
+        database.update_message_status(msg_id, "failed", "DAILY_SEND_LIMIT")
+        return False
     try:
         result = _worker.send_sms(to_number, body, timeout=120)
         status = "sent" if result["ok"] else "failed"
         database.update_message_status(msg_id, status, result.get("reason"))
+        return result["ok"]
     except OutboundQueueFullError:
         logger.warning("Internal outbound SMS rejected: QUEUE_FULL")
         database.update_message_status(msg_id, "failed", "QUEUE_FULL")
+        return False
     except WorkerStoppingError:
         logger.info("Internal outbound SMS rejected: SERVICE_STOPPING")
         database.update_message_status(msg_id, "failed", "SERVICE_STOPPING")
+        return False
     except Exception as e:
         logger.error("send_sms error: %s", e)
         database.update_message_status(msg_id, "failed", str(e))
+        return False
+
+    return False
 
 
 # ── App ───────────────────────────────────────────────────────────────────────
@@ -216,6 +249,18 @@ def validate_body(cls, v):
         return v.strip()
 
 
+class GrantPermissionRequest(BaseModel):
+    sapot_phone: str
+    external_phone: str
+
+    @field_validator("sapot_phone", "external_phone")
+    @classmethod
+    def validate_phone(cls, v):
+        if not re.match(r"^\+\d{7,15}$", v):
+            raise ValueError("phone must be E.164 format")
+        return v
+
+
 class AddUserRequest(BaseModel):
     phone:      str
     username:   str
@@ -267,11 +312,13 @@ async def health():
     )
 
 
-@app.get("/health/detailed", tags=["health"])
+@app.get(
+    "/health/detailed", tags=["health"], dependencies=[Depends(require_gsm_secret)]
+)
 def health_detailed(
     phone: Optional[str] = None,
         ):
-    """Full diagnostic — serial state, queue depth, pending SMS."""
+    """Full diagnostic -- serial state, queue depth, pending SMS."""
     if _worker is None:
         raise HTTPException(503, "Worker not initialised")
 
@@ -294,6 +341,8 @@ def pct(value):
         "connected":    _worker.connected,
         "last_status":  _worker.last_status,
         "queue_depth":  _worker.incoming_queue.qsize(),
+        "inbound_queue_capacity": _worker.incoming_queue.maxsize,
+        "inbound_queue_dropped": getattr(_worker, "incoming_queue_dropped", 0),
         "outbound_queue_depth": _worker.outbound_queue_depth,
         "outbound_queue_capacity": _worker.outbound_queue_capacity,
         "outbound_in_flight": _worker.outbound_in_flight,
@@ -329,7 +378,7 @@ def pct(value):
 
 # ── Status ────────────────────────────────────────────────────────────────────
 
-@app.get("/status", tags=["modem"])
+@app.get("/status", tags=["modem"], dependencies=[Depends(require_gsm_secret)])
 def status():
     """Modem and serial connection status."""
     if _worker is None:
@@ -365,6 +414,22 @@ def send_sms(req: SendSMSRequest):
         status="pending",
     )
 
+    if database.is_sms_opted_out(req.number):
+        database.update_message_status(msg_id, "failed", "RECIPIENT_OPTED_OUT")
+        raise HTTPException(403, {
+            "message": "Recipient has opted out of SMS relay messages",
+            "reason": "RECIPIENT_OPTED_OUT",
+            "msg_id": msg_id,
+        })
+
+    if not database.reserve_outbound_sms(settings.sms_daily_send_limit):
+        database.update_message_status(msg_id, "failed", "DAILY_SEND_LIMIT")
+        raise HTTPException(503, {
+            "message": "Daily SMS send limit reached",
+            "reason": "DAILY_SEND_LIMIT",
+            "msg_id": msg_id,
+        })
+
     try:
         result = _worker.send_sms(req.number, req.body, timeout=60)
     except OutboundQueueFullError:
@@ -408,7 +473,7 @@ def send_sms(req: SendSMSRequest):
     }
 
 
-@app.get("/sms/messages", tags=["sms"])
+@app.get("/sms/messages", tags=["sms"], dependencies=[Depends(require_gsm_secret)])
 def list_messages(
     limit:     int           = Query(50, ge=1, le=500),
     direction: Optional[str] = Query(None, pattern="^(IN|OUT)$"),
@@ -418,22 +483,51 @@ def list_messages(
     return database.get_messages(limit=limit, direction=direction, phone=phone)
 
 
+@app.post(
+    "/grant-permission",
+    tags=["permissions"],
+    dependencies=[Depends(require_gsm_secret)],
+)
+def grant_permission(req: GrantPermissionRequest):
+    """Record that a SAPOT user sent an outbound SMS to an external number.
+
+    Called by the main server after a confirmed /sms/send delivery.
+    """
+    database.grant_outbound_permission(req.sapot_phone, req.external_phone)
+    return {"ok": True}
+
+
+@app.get(
+    "/has-permission",
+    tags=["permissions"],
+    dependencies=[Depends(require_gsm_secret)],
+)
+def check_permission(
+    sapot_phone: str = Query(..., pattern=r"^\+\d{7,15}$"),
+    external_phone: str = Query(..., pattern=r"^\+\d{7,15}$"),
+):
+    """Return whether sapot_phone has an active outbound permission for external_phone."""
+    return {"permitted": database.has_outbound_permission(sapot_phone, external_phone)}
+
+
 # ── User endpoints ────────────────────────────────────────────────────────────
 
-@app.get("/users", tags=["users"])
+@app.get("/users", tags=["users"], dependencies=[Depends(require_gsm_secret)])
 def list_users():
     return database.get_all_users()
 
 
-@app.get("/users/{phone}", tags=["users"])
+@app.get("/users/{phone}", tags=["users"], dependencies=[Depends(require_gsm_secret)])
 def get_user(phone: str):
     user = database.lookup_number(phone)
     if not user:
         raise HTTPException(404, f"No user with phone {phone}")
     return user
 
 
-@app.post("/users", tags=["users"], status_code=201)
+@app.post(
+    "/users", tags=["users"], status_code=201, dependencies=[Depends(require_gsm_secret)]
+)
 def add_user(req: AddUserRequest):
     try:
         return database.add_user(req.phone, req.username, req.app_active)
@@ -445,15 +539,17 @@ def add_user(req: AddUserRequest):
 
 # ── Session endpoints ─────────────────────────────────────────────────────────
 
-@app.get("/sessions", tags=["sessions"])
+@app.get("/sessions", tags=["sessions"], dependencies=[Depends(require_gsm_secret)])
 def list_sessions():
     """All sessions currently in the database (for debugging)."""
     with database._conn() as cx:
         rows = cx.execute("SELECT * FROM sessions ORDER BY last_seen DESC").fetchall()
     return [dict(r) for r in rows]
 
 
-@app.delete("/sessions/{phone}", tags=["sessions"])
+@app.delete(
+    "/sessions/{phone}", tags=["sessions"], dependencies=[Depends(require_gsm_secret)]
+)
 def reset_session(phone: str):
     """Reset a user's conversation session back to NEW."""
     database.reset_session(phone)

GSM-module/GSM-fastapi/config.py

@@ -31,12 +31,29 @@ def bounded_integer_env(name: str, default: int, maximum: int) -> int:
     return parsed
 
 
+def nonnegative_integer_env(name: str, default: int) -> int:
+    value = os.environ.get(name)
+    if value is None:
+        return default
+    try:
+        parsed = int(value)
+    except ValueError as error:
+        raise RuntimeError(
+            f"Environment variable '{name}' must be a non-negative integer."
+        ) from error
+    if parsed < 0:
+        raise RuntimeError(
+            f"Environment variable '{name}' must be a non-negative integer."
+        )
+    return parsed
+
+
 class Settings:
     load_dotenv()
     # Serial port the Arduino is connected to
     serial_port: str = os.environ.get("SERIAL_PORT", "/dev/ttyACM0")
 
-    # Baud rate — must match PC_BAUD in the Arduino sketch (9600)
+    # Baud rate -- must match PC_BAUD in the Arduino sketch (9600)
     serial_baud: int = int(os.environ.get("SERIAL_BAUD", "9600"))
 
     db_path: str | None = os.environ.get("DB_PATH")
@@ -59,6 +76,20 @@ class Settings:
     sms_send_queue_maxsize: int = bounded_integer_env(
         "SMS_SEND_QUEUE_MAXSIZE", 10, MAX_SEND_QUEUE_SIZE
     )
+    sms_incoming_queue_maxsize: int = bounded_integer_env(
+        "SMS_INCOMING_QUEUE_MAXSIZE", 100, 10_000
+    )
+    sms_daily_send_limit: int = nonnegative_integer_env("SMS_DAILY_SEND_LIMIT", 100)
+    sms_sender_daily_limit: int = nonnegative_integer_env("SMS_SENDER_DAILY_LIMIT", 20)
+    sms_sender_target_daily_limit: int = nonnegative_integer_env(
+        "SMS_SENDER_TARGET_DAILY_LIMIT", 10
+    )
+    sms_response_cooldown_seconds: int = nonnegative_integer_env(
+        "SMS_RESPONSE_COOLDOWN_SECONDS", 30
+    )
+    sms_log_retention_days: int = nonnegative_integer_env("SMS_LOG_RETENTION_DAYS", 30)
+    log_max_bytes: int = bounded_integer_env("LOG_MAX_BYTES", 1_000_000, 100_000_000)
+    log_backup_count: int = bounded_integer_env("LOG_BACKUP_COUNT", 3, 100)
 
 
 settings = Settings()