test(security): cover legacy compatibility and route enforcement

Author: gsiri-code

security/tests/test_models_manager.py

@@ -0,0 +1,72 @@
+from security.manager import get_protocol, is_permitted, load_legacy_records
+from security.models import ActionChecks, SecProtocol
+from server.auth import ROLE_ADMIN, ROLE_USER, create_access_token
+
+
+def test_action_checks_enforces_login_users_and_roles():
+    checks = ActionChecks(login=True, valid_users=["alice"], allowed_roles=[ROLE_ADMIN])
+
+    assert checks.is_permitted("alice", {"role": ROLE_ADMIN})
+    assert not checks.is_permitted("", {"role": ROLE_ADMIN})
+    assert not checks.is_permitted("bob", {"role": ROLE_ADMIN})
+    assert not checks.is_permitted("alice", {"role": ROLE_USER})
+
+
+def test_protocol_round_trip_from_legacy_records():
+    records = {
+        "countries": {
+            "update": {
+                "user_list": ["alice"],
+                "checks": {
+                    "login": True,
+                    "allowed_roles": [ROLE_ADMIN],
+                },
+            },
+        },
+    }
+
+    load_legacy_records(records)
+    protocol = get_protocol("countries")
+
+    assert protocol is not None
+    assert protocol.name == "countries"
+    assert protocol.update.login is True
+    assert protocol.update.valid_users == ["alice"]
+    assert protocol.update.allowed_roles == [ROLE_ADMIN]
+
+
+def test_manager_is_permitted_uses_jwt_payload_when_no_user_id_is_passed():
+    records = {
+        "countries": {
+            "update": {
+                "user_list": ["alice"],
+                "checks": {
+                    "login": True,
+                    "allowed_roles": [ROLE_ADMIN],
+                },
+            },
+        },
+    }
+    load_legacy_records(records)
+    token = create_access_token("alice", ROLE_ADMIN, expires_hours=1)
+
+    assert is_permitted("countries", "update", auth_header=f"Bearer {token}")
+    assert not is_permitted("countries", "update")
+
+
+def test_unknown_protocol_is_open_by_default():
+    load_legacy_records({})
+    assert is_permitted("missing", "read")
+
+
+def test_duplicate_protocol_add_is_rejected():
+    protocol = SecProtocol("people")
+    load_legacy_records({})
+    from security.manager import add_protocol
+
+    add_protocol(protocol)
+    try:
+        add_protocol(protocol)
+        assert False, "Expected ValueError"
+    except ValueError as exc:
+        assert "Duplicate protocol" in str(exc)

security/tests/test_security.py

@@ -7,3 +7,17 @@ def test_read():
     for feature in recs:
         assert isinstance(feature, str)
         assert len(feature) > 0
+
+
+def test_read_feature_legacy_shape_is_preserved():
+    feature = sec.read_feature(sec.PEOPLE)
+    assert feature == sec.temp_recs[sec.PEOPLE]
+
+
+def test_read_protocol_builds_richer_protocol_view():
+    protocol = sec.read_protocol(sec.PEOPLE)
+    assert protocol is not None
+    assert protocol.name == sec.PEOPLE
+    assert protocol.create.login is True
+    assert sec.is_permitted(sec.PEOPLE, sec.CREATE, user_id='ejc369@nyu.edu')
+    assert not sec.is_permitted(sec.PEOPLE, sec.CREATE, user_id='someone-else@nyu.edu')

server/tests/test_security_decorators.py

@@ -0,0 +1,97 @@
+from flask import Flask
+
+from security.decorators import require_protocol
+from server.auth import ROLE_ADMIN, ROLE_USER, create_access_token
+
+
+def _create_test_app():
+    app = Flask(__name__)
+
+    @app.get("/protected")
+    @require_protocol("countries", "update")
+    def protected():
+        return {"ok": True}, 200
+
+    return app
+
+
+def test_security_decorator_disabled_by_default(monkeypatch):
+    monkeypatch.delenv("SECURITY_ENFORCEMENT", raising=False)
+    app = _create_test_app()
+
+    with app.test_client() as client:
+        response = client.get("/protected")
+
+    assert response.status_code == 200
+
+
+def test_security_decorator_blocks_when_enforcement_enabled(monkeypatch):
+    monkeypatch.setenv("SECURITY_ENFORCEMENT", "true")
+    app = _create_test_app()
+
+    from security.manager import load_legacy_records
+    load_legacy_records({
+        "countries": {
+            "update": {
+                "user_list": ["alice"],
+                "checks": {
+                    "login": True,
+                    "allowed_roles": [ROLE_ADMIN],
+                },
+            },
+        },
+    })
+
+    with app.test_client() as client:
+        response = client.get("/protected")
+
+    assert response.status_code == 403
+
+
+def test_security_decorator_allows_authorized_request(monkeypatch):
+    monkeypatch.setenv("SECURITY_ENFORCEMENT", "true")
+    app = _create_test_app()
+
+    from security.manager import load_legacy_records
+    load_legacy_records({
+        "countries": {
+            "update": {
+                "user_list": ["alice"],
+                "checks": {
+                    "login": True,
+                    "allowed_roles": [ROLE_ADMIN],
+                },
+            },
+        },
+    })
+    token = create_access_token("alice", ROLE_ADMIN, expires_hours=1)
+
+    with app.test_client() as client:
+        response = client.get("/protected", headers={"Authorization": f"Bearer {token}"})
+
+    assert response.status_code == 200
+
+
+def test_security_decorator_audit_only_does_not_block(monkeypatch):
+    monkeypatch.setenv("SECURITY_ENFORCEMENT", "true")
+    monkeypatch.setenv("SECURITY_AUDIT_ONLY", "true")
+    app = _create_test_app()
+
+    from security.manager import load_legacy_records
+    load_legacy_records({
+        "countries": {
+            "update": {
+                "user_list": ["alice"],
+                "checks": {
+                    "login": True,
+                    "allowed_roles": [ROLE_ADMIN],
+                },
+            },
+        },
+    })
+    token = create_access_token("alice", ROLE_USER, expires_hours=1)
+
+    with app.test_client() as client:
+        response = client.get("/protected", headers={"Authorization": f"Bearer {token}"})
+
+    assert response.status_code == 200