Add unit tests for API authentication flow

inventory_provider/auth.py

+from functools import wraps
 from flask import current_app, g, jsonify
 from flask_httpauth import HTTPTokenAuth
-from functools import wraps
-from config import ANONYMOUS_SERVICE_NAME
+
+from inventory_provider.config import ANONYMOUS_SERVICE_NAME
 
 auth = HTTPTokenAuth(scheme="ApiKey")
 
@@ -28,17 +29,11 @@ def authorize(*, allowed_clients):
         @wraps(f)
         def wrapped(*args, **kwargs):
             client = g.get("auth_client")
-
-            if not client:
-                return jsonify({"error": "Unauthorized"}), 403
-
             if client not in allowed_clients:
                 # Anonymous clients are allowed to access any resource without providing an API key
                 # TODO: Only for testing, should be removed in Production
                 if client != ANONYMOUS_SERVICE_NAME:
                     return jsonify({"error": "Forbidden"}), 403
-
             return f(*args, **kwargs)
-
         return wrapped
-    return decorator
\ No newline at end of file
+    return decorator

inventory_provider/routes/classifier.py

@@ -67,7 +67,8 @@ from redis import Redis
 from inventory_provider.routes import common
 from inventory_provider.routes.common import _ignore_cache_or_retrieve, cache_result
 
-from inventory_provider.config import authorize, DASHBOARD_SERVICE_NAME
+from inventory_provider.auth import authorize
+from inventory_provider.config import DASHBOARD_SERVICE_NAME
 
 routes = Blueprint("inventory-data-classifier-support-routes", __name__)
 

test/conftest.py

@@ -70,6 +70,17 @@ def data_config_filename():
 
     with tempfile.NamedTemporaryFile() as f:
         config = {
+            "api-keys": {
+                "brian": {
+                    "api-key": "brian_key"
+                },
+                "dashboard": {
+                    "api-key": "dashboard_key"
+                },
+                "reporting": {
+                    "api-key": "reporting_key"
+                },
+            },
             "ssh": {
                 "username": "uSeR-NaMe",
                 "private-key": "private-key-filename",

test/test_auth.py

+import jsonschema
+
+from inventory_provider.routes.classifier_schema import (
+    ROUTER_INFO_ALL_ROUTERS_RESPONSE_SCHEMA,
+)
+
+DEFAULT_REQUEST_HEADERS_NO_KEY = {
+    "Content-type": "application/json",
+    "Accept": ["application/json"],
+}
+
+DEFAULT_REQUEST_HEADERS_BRIAN_KEY = {
+    "Content-type": "application/json",
+    "Accept": ["application/json"],
+    "Authorization": "ApiKey brian_key",
+}
+
+DEFAULT_REQUEST_HEADERS_REPORTING_KEY = {
+    "Content-type": "application/json",
+    "Accept": ["application/json"],
+    "Authorization": "ApiKey reporting_key",
+}
+
+DEFAULT_REQUEST_HEADERS_DASHBOARD_KEY = {
+    "Content-type": "application/json",
+    "Accept": ["application/json"],
+    "Authorization": "ApiKey dashboard_key",
+}
+
+DEFAULT_REQUEST_HEADERS_BAD_KEY = {
+    "Content-type": "application/json",
+    "Accept": ["application/json"],
+    "Authorization": "ApiKey badapikey",
+}
+
+def test_classifier_router_no_key(client):
+    rv = client.get("/classifier/router-info", headers=DEFAULT_REQUEST_HEADERS_NO_KEY)
+    assert rv.status_code == 200
+    assert rv.is_json
+    result = rv.json
+
+    jsonschema.validate(result, ROUTER_INFO_ALL_ROUTERS_RESPONSE_SCHEMA)
+    assert len(result) > 0
+
+def test_classifier_router_dashboard_key(client):
+    rv = client.get("/classifier/router-info", headers=DEFAULT_REQUEST_HEADERS_DASHBOARD_KEY)
+    assert rv.status_code == 200
+    assert rv.is_json
+    result = rv.json
+
+    jsonschema.validate(result, ROUTER_INFO_ALL_ROUTERS_RESPONSE_SCHEMA)
+    assert len(result) > 0
+
+def test_classifier_router_brian_key(client):
+    rv = client.get("/classifier/router-info", headers=DEFAULT_REQUEST_HEADERS_BRIAN_KEY)
+    assert rv.status_code == 403
+    assert rv.is_json
+    result = rv.json
+
+    assert result["error"] == "Forbidden"
+
+def test_classifier_router_reporting_key(client):
+    rv = client.get("/classifier/router-info", headers=DEFAULT_REQUEST_HEADERS_REPORTING_KEY)
+    assert rv.status_code == 403
+    assert rv.is_json
+    result = rv.json
+
+    assert result["error"] == "Forbidden"
+
+def test_classifier_router_bad_key(client):
+    rv = client.get("/classifier/router-info", headers=DEFAULT_REQUEST_HEADERS_BAD_KEY)
+    assert rv.status_code == 401
+    result = rv.text
+
+    assert result == "Unauthorized Access"

test/test_flask_config.py

@@ -7,6 +7,17 @@ from inventory_provider.config import CONFIG_SCHEMA
 @pytest.fixture
 def config():
     return {
+        "api-keys": {
+            "brian": {
+                "api-key": "brian_key"
+            },
+            "dashboard": {
+                "api-key": "dashboard_key"
+            },
+            "reporting": {
+                "api-key": "reporting_key"
+            },
+        },
         'redis': {
             'hostname': 'localhost',
             'port': 6379,