feat(security): enforce countries protocol on write actions

Author: 818sushirolls

security/security.py

@@ -1,6 +1,8 @@
 from functools import wraps
 
 from security.manager import get_protocol, is_permitted as manager_is_permitted, load_legacy_records
+from security.models import ALLOWED_ROLES
+from server.auth import ROLE_ADMIN
 
 # import data.db_connect as dbc
 
@@ -61,6 +63,7 @@
 
 # Features:
 PEOPLE = 'people'
+COUNTRIES = 'countries'
 
 security_recs = None
 # These will come from the DB soon:
@@ -73,6 +76,26 @@
             },
         },
     },
+    COUNTRIES: {
+        CREATE: {
+            CHECKS: {
+                LOGIN: True,
+                ALLOWED_ROLES: [ROLE_ADMIN],
+            },
+        },
+        UPDATE: {
+            CHECKS: {
+                LOGIN: True,
+                ALLOWED_ROLES: [ROLE_ADMIN],
+            },
+        },
+        DELETE: {
+            CHECKS: {
+                LOGIN: True,
+                ALLOWED_ROLES: [ROLE_ADMIN],
+            },
+        },
+    },
 }
 
 

server/app.py

@@ -10,6 +10,7 @@
 from flask_restx import Api
 
 from data import db_connect
+from security.security import read as load_security_records
 
 APP_NAME = "Geographic Database API"
 APP_VERSION = "v1"
@@ -147,6 +148,8 @@ def create_app():
     if not any(isinstance(handler, InMemoryLogHandler) for handler in root_logger.handlers):
         root_logger.addHandler(InMemoryLogHandler())
 
+    load_security_records()
+
     cors_origins = get_runtime_cors_origins()
     CORS(app, resources={r"/*": {"origins": cors_origins}})
     api = Api(

server/countries_endpoints.py

@@ -10,6 +10,7 @@
 import data.countries as countries_data
 import data.states as states_data
 from data.countries import VALID_CONTINENTS
+from security import require_protocol
 from server.states_endpoints import state_model
 from server.helpers import apply_pagination, validate_pagination
 
@@ -307,6 +308,7 @@ def get(self):
                 HTTPStatus.INTERNAL_SERVER_ERROR, f"Database error: {str(e)}"
             )
 
+    @require_protocol("countries", "create")
     @countries_ns.doc("create_country")
     @countries_ns.expect(country_create_model)
     @countries_ns.marshal_with(country_model, code=HTTPStatus.CREATED)
@@ -436,6 +438,7 @@ def get(self, country_code):
                 f"Country with code '{country_code}' not found",
             )
 
+    @require_protocol("countries", "update")
     @countries_ns.doc("update_country")
     @countries_ns.expect(country_update_model)
     @countries_ns.marshal_with(country_model)
@@ -482,6 +485,7 @@ def put(self, country_code):
                 f"Country with code '{country_code}' not found",
             )
 
+    @require_protocol("countries", "delete")
     @countries_ns.doc("delete_country")
     @countries_ns.response(HTTPStatus.NO_CONTENT, "Country deleted successfully")
     @countries_ns.response(HTTPStatus.NOT_FOUND, "Country not found", error_model)

server/tests/test_countries_security.py

@@ -0,0 +1,110 @@
+"""Tests for the security protocol enforcement on countries write actions."""
+import json
+from http import HTTPStatus
+from unittest.mock import patch
+
+import pytest
+
+from server.app import create_app
+from server.auth import ROLE_ADMIN, ROLE_USER, create_access_token
+
+
+SAMPLE_COUNTRY = {
+    "country_name": "Test Country",
+    "country_code": "TC",
+    "continent": "North America",
+    "capital": "Test Capital",
+    "population": 1000000,
+    "area_km2": 50000.0,
+}
+
+
+@pytest.fixture
+def enforced_client(monkeypatch):
+    monkeypatch.setenv("SECURITY_ENFORCEMENT", "true")
+    monkeypatch.delenv("SECURITY_AUDIT_ONLY", raising=False)
+    app = create_app()
+    app.config["TESTING"] = True
+    with app.test_client() as client:
+        yield client
+
+
+def _bearer(role: str, user_id: str = "alice") -> dict:
+    token = create_access_token(user_id, role, expires_hours=1)
+    return {"Authorization": f"Bearer {token}"}
+
+
+def test_post_country_without_token_is_forbidden(enforced_client):
+    response = enforced_client.post(
+        "/countries",
+        data=json.dumps(SAMPLE_COUNTRY),
+        content_type="application/json",
+    )
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+def test_post_country_with_user_role_is_forbidden(enforced_client):
+    response = enforced_client.post(
+        "/countries",
+        data=json.dumps(SAMPLE_COUNTRY),
+        content_type="application/json",
+        headers=_bearer(ROLE_USER),
+    )
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+def test_post_country_with_admin_role_is_allowed(enforced_client):
+    with patch("data.countries.add_country") as mock_add:
+        mock_add.return_value = True
+        response = enforced_client.post(
+            "/countries",
+            data=json.dumps(SAMPLE_COUNTRY),
+            content_type="application/json",
+            headers=_bearer(ROLE_ADMIN),
+        )
+    assert response.status_code == HTTPStatus.CREATED
+
+
+def test_put_country_without_token_is_forbidden(enforced_client):
+    response = enforced_client.put(
+        "/countries/US",
+        data=json.dumps({"population": 350000000}),
+        content_type="application/json",
+    )
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+def test_put_country_with_admin_role_is_allowed(enforced_client):
+    with patch("data.countries.update_country") as mock_update, \
+            patch("data.countries.get_country_by_code") as mock_get:
+        mock_update.return_value = True
+        mock_get.return_value = SAMPLE_COUNTRY
+        response = enforced_client.put(
+            "/countries/TC",
+            data=json.dumps({"population": 2000000}),
+            content_type="application/json",
+            headers=_bearer(ROLE_ADMIN),
+        )
+    assert response.status_code == HTTPStatus.OK
+
+
+def test_delete_country_without_token_is_forbidden(enforced_client):
+    response = enforced_client.delete("/countries/US")
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+def test_delete_country_with_admin_role_is_allowed(enforced_client):
+    with patch("data.countries.delete_country") as mock_delete:
+        mock_delete.return_value = True
+        response = enforced_client.delete(
+            "/countries/TC",
+            headers=_bearer(ROLE_ADMIN),
+        )
+    assert response.status_code == HTTPStatus.NO_CONTENT
+
+
+def test_get_countries_remains_open_under_enforcement(enforced_client):
+    with patch("data.countries.get_countries_filtered") as mock_get:
+        mock_get.return_value = []
+        response = enforced_client.get("/countries")
+    assert response.status_code == HTTPStatus.OK