feat: harden data access and simplify user management

tests/test_security_baseline.py

@@ -10,7 +10,7 @@ from app.auth import (
     create_access_token,
     get_current_user,
     require_admin,
-    require_data_operator,
+    require_data_viewer,
 )
 from app.config import settings
 from app.data_platform import interface_service
@@ -28,8 +28,17 @@ def _production_settings(**updates):
         "database_url": "postgresql://app:" + "p" * 32 + "@127.0.0.1:5432/kg",
         "data_mysql_url": "mysql://app:" + "m" * 32 + "@127.0.0.1:3306/control",
         "mysql_host_bind": "127.0.0.1",
+        "data_mysql_storage_backend": "dedicated-block",
+        "data_mysql_storage_id": "prod-mysql-storage-01",
+        "data_mysql_storage_mount": "/mnt/sdr",
+        "data_mysql_data_dir": "/mnt/sdr/nianxx/mysql",
+        "data_backup_enabled": True,
+        "data_backup_encryption_required": True,
+        "data_backup_retention_days": 30,
+        "data_backup_root": "/mnt/backup/nianxx/mysql",
         "data_mysql_direct_access_enabled": True,
         "data_mysql_ssh_tunnel_required": True,
+        "data_mysql_ssh_host": "db-admin.example.cn",
         "cors_allowed_origins": "https://data.example.cn",
         "trusted_hosts": "data.example.cn",
         "auth_algorithm": "HS256",
@@ -69,6 +78,68 @@ class SecurityBaselineTests(unittest.TestCase):
             {item["code"] for item in report["findings"]},
         )
 
+    def test_production_rejects_project_local_or_unidentified_mysql_storage(self) -> None:
+        unsafe = _production_settings(
+            data_mysql_storage_backend="docker-volume",
+            data_mysql_storage_id="",
+            data_mysql_data_dir="/app/mysql-data",
+        )
+        codes = {item["code"] for item in security_readiness(unsafe)["findings"]}
+        self.assertIn("MYSQL_STORAGE_BACKEND", codes)
+
+        unsafe = _production_settings(
+            data_mysql_storage_id="short",
+            data_mysql_storage_mount="/app/mysql-storage",
+            data_mysql_data_dir="/app/mysql-data",
+        )
+        codes = {item["code"] for item in security_readiness(unsafe)["findings"]}
+        self.assertIn("MYSQL_STORAGE_ID", codes)
+        self.assertIn("MYSQL_STORAGE_MOUNT", codes)
+        self.assertIn("MYSQL_DATA_DIR", codes)
+
+        outside = _production_settings(data_mysql_data_dir="/srv/mysql")
+        self.assertIn(
+            "MYSQL_DATA_OUTSIDE_MOUNT",
+            {item["code"] for item in security_readiness(outside)["findings"]},
+        )
+
+    def test_production_requires_encrypted_external_nonoverlapping_backups(self) -> None:
+        unsafe = _production_settings(
+            data_backup_enabled=False,
+            data_backup_encryption_required=False,
+            data_backup_root="/mnt/sdr/nianxx/mysql/backups",
+        )
+        codes = {item["code"] for item in security_readiness(unsafe)["findings"]}
+        self.assertIn("BACKUP_DISABLED", codes)
+
+        unsafe = _production_settings(
+            data_backup_encryption_required=False,
+            data_backup_retention_days=1,
+            data_backup_root="/mnt/sdr/nianxx/mysql/backups",
+        )
+        codes = {item["code"] for item in security_readiness(unsafe)["findings"]}
+        self.assertIn("BACKUP_ENCRYPTION_DISABLED", codes)
+        self.assertIn("BACKUP_RETENTION", codes)
+        self.assertIn("BACKUP_DATA_OVERLAP", codes)
+
+    def test_managed_dbeaver_access_requires_an_isolated_provisioner(self) -> None:
+        safe = _production_settings(
+            data_mysql_managed_access_enabled=True,
+            data_mysql_ssh_authorized_keys_file="/srv/dbeaver/authorized_keys",
+            data_mysql_ssh_host_public_key_file="/srv/dbeaver/host.pub",
+            data_mysql_provisioner_user="access_broker",
+            data_mysql_provisioner_password="z" * 32,
+        )
+        self.assertEqual(security_readiness(safe)["critical_count"], 0)
+
+        unsafe = safe.model_copy(update={
+            "data_mysql_provisioner_user": "app",
+            "data_mysql_provisioner_password": "change-me",
+        })
+        codes = {item["code"] for item in security_readiness(unsafe)["findings"]}
+        self.assertIn("DBEAVER_PROVISIONER_USER", codes)
+        self.assertIn("DBEAVER_PROVISIONER_PASSWORD", codes)
+
 
 class AuthenticationSecurityTests(unittest.IsolatedAsyncioTestCase):
     async def test_token_uses_issuer_audience_and_live_database_roles(self) -> None:
@@ -121,14 +192,22 @@ class AuthenticationSecurityTests(unittest.IsolatedAsyncioTestCase):
             require_admin({"username": "collector", "roles": ["collector"]})
         self.assertEqual(raised.exception.status_code, 403)
 
-    def test_data_center_rejects_collector_role(self) -> None:
+    def test_data_center_view_permission_rejects_unassigned_accounts(self) -> None:
         with self.assertRaises(HTTPException) as raised:
-            require_data_operator({"username": "collector", "roles": ["collector"]})
+            require_data_viewer({"username": "unassigned", "roles": []})
         self.assertEqual(raised.exception.status_code, 403)
-        operator = require_data_operator(
-            {"username": "operator", "roles": ["operator"]}
+        viewer = require_data_viewer(
+            {"username": "viewer", "roles": ["operator"]}
         )
-        self.assertEqual(operator["username"], "operator")
+        self.assertEqual(viewer["username"], "viewer")
+
+    def test_retired_collector_accounts_are_read_only_compatible(self) -> None:
+        legacy = require_data_viewer(
+            {"username": "legacy", "roles": ["collector"]}
+        )
+        self.assertEqual(legacy["username"], "legacy")
+        with self.assertRaises(HTTPException):
+            require_admin(legacy)
 
 
 class InterfaceCredentialSecurityTests(unittest.TestCase):