feat(safety+audit+deploy): consistency_check + audit log file sink + deploy script

#2 Env switch safety:
- mcp_common/environment.py: nuova consistency_check() che previene
  switch accidentali a mainnet. Solleva EnvironmentMismatchError se
  resolved=mainnet senza creds["environment"]="mainnet" esplicito,
  o se declared/resolved mismatch. Override via STRICT_MAINNET=false.
- Wirato in app_factory.run_exchange_main al boot.
- 6 nuovi test consistency.

#3 Audit log persistence:
- mcp_common/audit.py: TimedRotatingFileHandler aggiuntivo se env
  AUDIT_LOG_FILE settato. Rotation midnight UTC, retention 30gg
  default (AUDIT_LOG_BACKUP_DAYS). Format JSONL con SecretsFilter.
- docker-compose.prod.yml: bind mount /var/log/cerbero-mcp + env
  AUDIT_LOG_FILE per i 4 servizi exchange (write endpoints).
- 2 nuovi test file sink.

#1 Deploy script:
- scripts/deploy.sh: idempotente, fa docker login + clone/pull repo +
  copia secrets chmod 600 + crea .env + setup audit dir + pull image
  + up + smoke test pubblico HTTPS.
- DEPLOYMENT.md aggiornato: sezioni 2 (script), 3 (safety mainnet),
  4 (audit log query), renumber sezioni successive.

Test: 488/488 verdi.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

services/common/src/mcp_common/app_factory.py

@@ -24,7 +24,11 @@ import uvicorn
 
 from mcp_common.auth import load_token_store_from_files
 from mcp_common.env_validation import fail_fast_if_missing, require_env, summarize
-from mcp_common.environment import EnvironmentInfo, resolve_environment
+from mcp_common.environment import (
+    EnvironmentInfo,
+    consistency_check,
+    resolve_environment,
+)
 from mcp_common.logging import configure_root_logging
 
 
@@ -69,6 +73,11 @@ def run_exchange_main(spec: ExchangeAppSpec) -> None:
         default_base_url_testnet=spec.default_base_url_testnet,
     )
 
+    # Safety: previene switch accidentali a mainnet senza conferma esplicita
+    # nel secret. Solleva EnvironmentMismatchError → boot abort se mismatch.
+    strict_mainnet = os.environ.get("STRICT_MAINNET", "true").lower() not in ("0", "false", "no")
+    consistency_check(env_info, creds, strict_mainnet=strict_mainnet)
+
     client = spec.build_client(creds, env_info)
 
     token_store = load_token_store_from_files(

services/common/src/mcp_common/audit.py

@@ -1,22 +1,68 @@
 """Audit log strutturato per write endpoint MCP (place_order, cancel,
 set_*, close_*, transfer_*). Usa un logger dedicato `mcp.audit` su stream
-JSON: in deployment può essere redirezionato a file/syslog/SIEM separato.
+JSON.
 
-Logica:
-- `audit_write_op(principal, action, exchange, target, payload, result)`
-  emette UN record JSON per ogni operazione con esito (ok/error).
-- Payload sensibile (api_key, secret) già filtrato dal SecretsFilter
-  globale; qui non si include creds.
+Sink:
+- stdout/stderr (sempre): tramite root JSON logger configurato da
+  `mcp_common.logging.configure_root_logging`.
+- File JSONL persistente (opzionale): se env var `AUDIT_LOG_FILE` è
+  settata, aggiunge un `TimedRotatingFileHandler` che ruota a mezzanotte
+  con `AUDIT_LOG_BACKUP_DAYS` di retention (default 30). Una riga JSON
+  per record (formato `.jsonl`).
+
+Per VPS produzione: setta `AUDIT_LOG_FILE=/var/log/cerbero-mcp/<service>.audit.jsonl`
+con bind mount del volume `/var/log/cerbero-mcp` nel docker-compose.
+
+Payload sensibile (api_key, secret) già filtrato dal SecretsFilter
+globale; qui non si include creds.
 """
 from __future__ import annotations
 
 import logging
+import os
+from logging.handlers import TimedRotatingFileHandler
 from typing import Any
 
 from mcp_common.auth import Principal
-from mcp_common.logging import get_json_logger
+from mcp_common.logging import SecretsFilter, get_json_logger
+
+try:
+    from pythonjsonlogger.json import JsonFormatter as _JsonFormatter  # noqa: N813
+except ImportError:
+    from pythonjsonlogger.jsonlogger import JsonFormatter as _JsonFormatter  # noqa: N813
 
 _logger = get_json_logger("mcp.audit", level=logging.INFO)
+_file_handler_attached = False
+
+
+def _configure_audit_sink() -> None:
+    """Aggiunge FileHandler al logger mcp.audit se AUDIT_LOG_FILE è settato.
+    Idempotente: chiamato la prima volta da audit_write_op, poi no-op.
+    """
+    global _file_handler_attached
+    if _file_handler_attached:
+        return
+
+    file_path = os.environ.get("AUDIT_LOG_FILE", "").strip()
+    if not file_path:
+        _file_handler_attached = True
+        return
+
+    backup_days = int(os.environ.get("AUDIT_LOG_BACKUP_DAYS", "30"))
+
+    os.makedirs(os.path.dirname(file_path) or ".", exist_ok=True)
+    handler = TimedRotatingFileHandler(
+        file_path,
+        when="midnight",
+        interval=1,
+        backupCount=backup_days,
+        encoding="utf-8",
+        utc=True,
+    )
+    handler.setFormatter(_JsonFormatter("%(asctime)s %(name)s %(levelname)s %(message)s"))
+    handler.addFilter(SecretsFilter())
+    _logger.addHandler(handler)
+    _file_handler_attached = True
 
 
 def audit_write_op(
@@ -40,6 +86,7 @@ def audit_write_op(
     result: output del client (order_id, status, ecc.).
     error: stringa errore se l'operazione ha fallito.
     """
+    _configure_audit_sink()
     record: dict[str, Any] = {
         "audit_event": "write_op",
         "action": action,

services/common/src/mcp_common/environment.py

@@ -1,18 +1,31 @@
 """Resolver di ambiente (testnet/mainnet) per MCP exchange.
 
 Precedenza: env var > campo secret > default True (testnet).
+
+Safety: `consistency_check` previene switch accidentali a mainnet senza
+conferma esplicita nel secret JSON.
 """
 from __future__ import annotations
 
+import logging
 import os
 from dataclasses import dataclass
 from typing import Literal
 
+logger = logging.getLogger(__name__)
+
 Environment = Literal["testnet", "mainnet"]
 Source = Literal["env", "credentials", "default"]
 
 TRUTHY = {"1", "true", "yes", "on"}
 
+# Tokens nel base_url che indicano endpoint testnet (case-insensitive).
+TESTNET_URL_HINTS = ("test", "testnet", "paper")
+
+
+class EnvironmentMismatchError(RuntimeError):
+    """Boot abort: ambiente risolto non matcha conferma esplicita nel secret."""
+
 
 @dataclass(frozen=True)
 class EnvironmentInfo:
@@ -67,3 +80,59 @@ def resolve_environment(
         env_value=env_value,
         base_url=base_url,
     )
+
+
+def consistency_check(
+    env_info: EnvironmentInfo,
+    creds: dict,
+    *,
+    strict_mainnet: bool = True,
+) -> list[str]:
+    """Verifica coerenza environment risolto vs secret JSON. Restituisce
+    lista di warning string. Solleva EnvironmentMismatchError per mismatch
+    bloccanti.
+
+    Regole:
+    - Se `creds["environment"]` è presente e DIVERSO da `env_info.environment`:
+      → raise EnvironmentMismatchError (declared vs resolved mismatch).
+    - Se `env_info.environment == "mainnet"` e `creds.get("environment") !=
+      "mainnet"`: con `strict_mainnet=True` → raise (richiede conferma
+      esplicita). Con `strict_mainnet=False` → warning.
+    - Se `env_info.base_url` contiene token testnet ("test", "testnet",
+      "paper") ma `env_info.environment == "mainnet"` (o viceversa): warning
+      (URL/environment incoerenti).
+    """
+    warnings: list[str] = []
+
+    declared = creds.get("environment")
+    if declared and declared != env_info.environment:
+        raise EnvironmentMismatchError(
+            f"{env_info.exchange}: secret declared environment={declared!r} "
+            f"but resolver resolved environment={env_info.environment!r}"
+        )
+
+    if env_info.environment == "mainnet" and declared != "mainnet":
+        msg = (
+            f"{env_info.exchange}: resolved mainnet without explicit confirmation "
+            "in secret. Add `\"environment\": \"mainnet\"` to the credentials JSON."
+        )
+        if strict_mainnet:
+            raise EnvironmentMismatchError(msg)
+        warnings.append(msg)
+
+    url_lower = (env_info.base_url or "").lower()
+    has_test_hint = any(token in url_lower for token in TESTNET_URL_HINTS)
+    if env_info.environment == "mainnet" and has_test_hint:
+        warnings.append(
+            f"{env_info.exchange}: environment=mainnet but base_url contains "
+            f"testnet hint ({env_info.base_url!r})"
+        )
+    if env_info.environment == "testnet" and not has_test_hint and url_lower:
+        warnings.append(
+            f"{env_info.exchange}: environment=testnet but base_url does not "
+            f"appear to be a testnet endpoint ({env_info.base_url!r})"
+        )
+
+    for w in warnings:
+        logger.warning("environment consistency: %s", w)
+    return warnings