GH-4: Require authorized credentials for API

schemes/basic_auth.py

@@ -17,6 +17,13 @@ def authorized(auth: Authorization | None) -> bool:
     return basic_auth(authorized)
 
 
+def api() -> Callable[[Callable[P, T]], Callable[P, T | Response]]:
+    def authorized(auth: Authorization | None) -> bool:
+        return _authorized(auth, current_app.config["API_USERNAME"], current_app.config["API_PASSWORD"])
+
+    return basic_auth(authorized)
+
+
 def basic_auth(authorized: Authorized) -> Callable[[Callable[P, T]], Callable[P, T | Response]]:
     def decorator(func: Callable[P, T]) -> Callable[P, T | Response]:
         @functools.wraps(func)

schemes/users.py

@@ -5,6 +5,8 @@
 from flask import Blueprint, Response, request
 from sqlalchemy import Column, Engine, Integer, MetaData, String, Table, text
 
+from schemes import basic_auth
+
 
 @dataclass
 class User:
@@ -64,6 +66,7 @@ def get_all(self) -> List[User]:
 
 
 @bp.route("", methods=["POST"])
+@basic_auth.api()
 @inject.autoparams()
 def add(users: UserRepository) -> Response:
     json = request.get_json()
@@ -72,6 +75,7 @@ def add(users: UserRepository) -> Response:
 
 
 @bp.route("", methods=["DELETE"])
+@basic_auth.api()
 @inject.autoparams()
 def clear(users: UserRepository) -> Response:
     users.clear()

tests/e2e/app_client.py

@@ -1,17 +1,28 @@
+from base64 import b64encode
+
 import requests
 
 
 class AppClient:
     DEFAULT_TIMEOUT = 10
 
-    def __init__(self, url: str):
+    def __init__(self, url: str, username: str, password: str):
         self._url = url
+        credentials = f"{username}:{password}"
+        self._authorization = f"Basic {b64encode(credentials.encode()).decode()}"
 
     def add_user(self, email: str) -> None:
         users = [{"email": email}]
-        response = requests.post(f"{self._url}/users", json=users, timeout=self.DEFAULT_TIMEOUT)
+        response = requests.post(
+            f"{self._url}/users",
+            headers={"Authorization": self._authorization},
+            json=users,
+            timeout=self.DEFAULT_TIMEOUT,
+        )
         assert response.status_code == 201
 
     def clear_users(self) -> None:
-        response = requests.delete(f"{self._url}/users", timeout=self.DEFAULT_TIMEOUT)
+        response = requests.delete(
+            f"{self._url}/users", headers={"Authorization": self._authorization}, timeout=self.DEFAULT_TIMEOUT
+        )
         assert response.status_code == 204

tests/e2e/conftest.py

@@ -32,9 +32,15 @@ def configure_live_server_fixture() -> None:
         multiprocessing.set_start_method("fork")
 
 
+@pytest.fixture(name="api_credentials", scope="package")
+def api_credentials_fixture() -> tuple[str, str]:
+    return "api-user", "api-password"
+
+
 @pytest.fixture(name="app", scope="package")
-def app_fixture(oidc_server: LiveServer) -> Generator[Flask, Any, Any]:
+def app_fixture(api_credentials: tuple[str, str], oidc_server: LiveServer) -> Generator[Flask, Any, Any]:
     port = _get_random_port()
+    api_username, api_password = api_credentials
     client_id = "app"
     private_key, public_key = _generate_key_pair()
 
@@ -44,6 +50,8 @@ def app_fixture(oidc_server: LiveServer) -> Generator[Flask, Any, Any]:
             "SECRET_KEY": b"secret_key",
             "SERVER_NAME": f"localhost:{port}",
             "LIVESERVER_PORT": port,
+            "API_USERNAME": api_username,
+            "API_PASSWORD": api_password,
             "GOVUK_CLIENT_ID": client_id,
             "GOVUK_CLIENT_SECRET": private_key.decode(),
             "GOVUK_SERVER_METADATA_URL": oidc_server.app.url_for("openid_configuration", _external=True),
@@ -67,8 +75,9 @@ def app_fixture(oidc_server: LiveServer) -> Generator[Flask, Any, Any]:
 
 
 @pytest.fixture(name="app_client")
-def app_client_fixture(live_server: LiveServer) -> Generator[AppClient, Any, Any]:
-    client = AppClient(_get_url(live_server))
+def app_client_fixture(live_server: LiveServer, api_credentials: tuple[str, str]) -> Generator[AppClient, Any, Any]:
+    username, password = api_credentials
+    client = AppClient(_get_url(live_server), username, password)
     yield client
     client.clear_users()
 

tests/integration/test_users.py

@@ -6,28 +6,72 @@
 
 from schemes.users import User, UserRepository
 
+BOARDMAN = "Basic Ym9hcmRtYW46bGV0bWVpbg=="  # echo -n 'boardman:letmein' | base64
+OBREE = "Basic b2JyZWU6b3BlbnNlc2FtZQ=="  # echo -n 'obree:opensesame' | base64
+
+
+@pytest.fixture(name="config")
+def config_fixture(config: Mapping[str, Any]) -> Mapping[str, Any]:
+    return config | {"API_USERNAME": "boardman", "API_PASSWORD": "letmein"}
+
 
 @pytest.fixture(name="users")
 def users_fixture() -> UserRepository:
     return inject.instance(UserRepository)
 
 
 def test_add_users(users: UserRepository, client: FlaskClient) -> None:
-    response = client.post("/users", json=[{"email": "boardman@example.com"}, {"email": "obree@example.com"}])
+    response = client.post(
+        "/users",
+        headers={"Authorization": BOARDMAN},
+        json=[{"email": "boardman@example.com"}, {"email": "obree@example.com"}],
+    )
 
     assert response.status_code == 201
     assert users.get_all() == [User("boardman@example.com"), User("obree@example.com")]
 
 
+def test_cannot_add_users_when_no_credentials(users: UserRepository, client: FlaskClient) -> None:
+    response = client.post("/users", json=[{"email": "boardman@example.com"}])
+
+    assert response.status_code == 401
+    assert not users.get_all()
+
+
+def test_cannot_add_users_when_incorrect_credentials(users: UserRepository, client: FlaskClient) -> None:
+    response = client.post("/users", headers={"Authorization": OBREE}, json=[{"email": "boardman@example.com"}])
+
+    assert response.status_code == 401
+    assert not users.get_all()
+
+
 def test_clear_users(users: UserRepository, client: FlaskClient) -> None:
     users.add(User("boardman@example.com"))
 
-    response = client.delete("/users")
+    response = client.delete("/users", headers={"Authorization": BOARDMAN})
 
     assert response.status_code == 204
     assert not users.get_all()
 
 
+def test_cannot_clear_users_when_no_credentials(users: UserRepository, client: FlaskClient) -> None:
+    users.add(User("boardman@example.com"))
+
+    response = client.delete("/users")
+
+    assert response.status_code == 401
+    assert users.get_all() == [User("boardman@example.com")]
+
+
+def test_cannot_clear_users_when_incorrect_credentials(users: UserRepository, client: FlaskClient) -> None:
+    users.add(User("boardman@example.com"))
+
+    response = client.delete("/users", headers={"Authorization": OBREE})
+
+    assert response.status_code == 401
+    assert users.get_all() == [User("boardman@example.com")]
+
+
 class TestProduction:
     @pytest.fixture(name="config")
     def config_fixture(self, config: Mapping[str, Any]) -> Mapping[str, Any]: