Merge pull request #16 from City-of-Helsinki/feature/add-cesvanoise

Feature/add cesvanoise

endpoints/__init__.py

@@ -46,31 +46,6 @@ def log_match(header: str, ip: str, allowed_network):
     return False
 
 
-
-# TODO: remove as this is not used anywhere
-class BaseRequestHandler(abc.ABC):
-    def __init__(self):
-        pass
-
-    @abc.abstractmethod
-    def process_request(self, request_data: dict) -> Tuple[bool, Union[str, None], Union[str, dict, list], int]:
-        """
-        Validate request and generate response
-        :param request_data:
-        :return: (
-            bool: request was valid or not
-            str: kafka topic's name
-            str/dict/list: response str or dict
-            int: HTTP status code
-        )
-        """
-        auth_ok = True
-        topic_name = os.getenv("KAFKA_RAW_DATA_TOPIC_NAME")
-        response_message = {"status": "ok"}
-        status_code = 200
-        return auth_ok, topic_name, response_message, status_code
-
-
 class AsyncRequestHandler(abc.ABC):
     """
     Async version of BaseRequestHandler, compatible with Starlette, FastAPI and Device registry.
@@ -79,6 +54,40 @@ class AsyncRequestHandler(abc.ABC):
     def __init__(self):
         pass
 
+    async def validate(
+        self, request_data: dict, endpoint_data: dict
+    ) -> Tuple[bool, Union[str, None], Union[int, None]]:
+        """
+        Use Starlette request_data here to determine should we accept or reject
+        this request
+        :param request_data: deserialized (FastAPI) Starlette Request
+        :param endpoint_data: endpoint data from device registry
+        :return: (bool ok, str error text, int status code)
+        """
+        # Reject requests not matching the one defined in env
+        if request_data["path"] != endpoint_data["endpoint_path"]:
+            return False, "Not found", 404
+        # Reject requests without token parameter, which can be in query string or http header
+        api_key = request_data["request"]["get"].get("x-api-key")
+        if api_key is None:
+            api_key = request_data["request"]["headers"].get("x-api-key")
+        if api_key is None or api_key != endpoint_data["auth_token"]:
+            logging.warning("Missing or invalid authentication token (x-api-key)")
+            return False, "Missing or invalid authentication token, see logs for error", 401
+        logging.info("Authentication token validated")
+        if request_data["request"]["get"].get("test") == "true":
+            logging.info("Test ok")
+            return False, "Test OK", 400
+        allowed_ip_addresses = endpoint_data.get("allowed_ip_addresses", "")
+        if allowed_ip_addresses == "":
+            logging.warning("Set 'allowed_ip_addresses' in endpoint settings to restrict requests unknown sources")
+        else:
+            if is_ip_address_allowed(request_data, allowed_ip_addresses) is False:
+                return False, "IP address not allowed", 403
+
+        # if all checks passed, return True
+        return True, None, None
+
     @abc.abstractmethod
     async def process_request(
         self, request_data: dict, endpoint_data: dict

endpoints/default/apikeyauth.py

@@ -1,13 +1,13 @@
-import logging
 from pprint import pprint
 from typing import Tuple, Union
 
-from .. import AsyncRequestHandler, is_ip_address_allowed
+from .. import AsyncRequestHandler
 
 
 class RequestHandler(AsyncRequestHandler):
-    @staticmethod
-    async def validate(request_data: dict, endpoint_data: dict) -> Tuple[bool, Union[str, None], Union[int, None]]:
+    async def validate(
+        self, request_data: dict, endpoint_data: dict
+    ) -> Tuple[bool, Union[str, None], Union[int, None]]:
         """
         Use Starlette request_data and endpoint_data from the Device registry here to determine
         should we accept or reject this request.
@@ -18,28 +18,7 @@ async def validate(request_data: dict, endpoint_data: dict) -> Tuple[bool, Union
         """
         pprint(request_data)
         pprint(endpoint_data)
-        # Reject requests not matching the one defined in env
-        # Compare request's full path with endpoint's path
-        if request_data["path"] != endpoint_data["endpoint_path"]:
-            return False, "Not found", 404
-        # Reject requests without token parameter, which can be in query string or http header
-        api_key = request_data["request"]["get"].get("x-api-key")
-        if api_key is None:
-            api_key = request_data["request"]["headers"].get("x-api-key")
-        if api_key is None or api_key != endpoint_data["auth_token"]:
-            logging.warning("Missing or invalid authentication token (x-api-key)")
-            return False, "Missing or invalid authentication token, see logs for error", 401
-        elif request_data["request"]["get"].get("test") == "true":
-            logging.info("Test ok")
-            return False, "Test OK", 400
-        # Reject requests from unknown IP addresses
-        allowed_ip_addresses = endpoint_data.get("allowed_ip_addresses", "")
-        if allowed_ip_addresses == "":
-            logging.warning("Set 'allowed_ip_addresses' in endpoint settings to restrict requests from unknown sources")
-        else:
-            if is_ip_address_allowed(request_data, allowed_ip_addresses) is False:
-                return False, "IP address not allowed", 403
-        return True, None, None
+        return super().validate(request_data, endpoint_data)
 
     async def process_request(
         self, request_data: dict, endpoint_data: dict

endpoints/digita/aiothingpark.py

@@ -2,39 +2,24 @@
 import os
 from typing import Tuple, Union
 
-from .. import AsyncRequestHandler, is_ip_address_allowed
+from .. import AsyncRequestHandler
 
 
 class RequestHandler(AsyncRequestHandler):
-    @staticmethod
-    async def validate(request_data: dict, endpoint_data: dict) -> Tuple[bool, Union[str, None], Union[int, None]]:
+    async def validate(
+        self, request_data: dict, endpoint_data: dict
+    ) -> Tuple[bool, Union[str, None], Union[int, None]]:
         """
         Use Starlette request_data here to determine should we accept or reject
         this request
         :param request_data: deserialized (FastAPI) Starlette Request
         :param endpoint_data: endpoint data from device registry
         :return: (bool ok, str error text, int status code)
         """
-        # Reject requests not matching the one defined in env
-        if request_data["path"] != endpoint_data["endpoint_path"]:
-            return False, "Not found", 404
-        # Reject requests without token parameter, which can be in query string or http header
-        api_key = request_data["request"]["get"].get("x-api-key")
-        if api_key is None:
-            api_key = request_data["request"]["headers"].get("x-api-key")
-        if api_key is None or api_key != endpoint_data["auth_token"]:
-            logging.warning("Missing or invalid authentication token (x-api-key)")
-            return False, "Missing or invalid authentication token, see logs for error", 401
-        logging.info("Authentication token validated")
-        if request_data["request"]["get"].get("test") == "true":
-            logging.info("Test ok")
-            return False, "Test OK", 400
-        allowed_ip_addresses = endpoint_data.get("allowed_ip_addresses", "")
-        if allowed_ip_addresses == "":
-            logging.warning("Set 'allowed_ip_addresses' in endpoint settings to restrict requests unknown sources")
-        else:
-            if is_ip_address_allowed(request_data, allowed_ip_addresses) is False:
-                return False, "IP address not allowed", 403
+        [status_ok, response_message, status_code] = await super().validate(request_data, endpoint_data)
+
+        if status_ok is False:
+            return False, response_message, status_code
 
         if request_data["request"]["get"].get("LrnDevEui") is None:
             logging.warning("LrnDevEui not found in request params")

endpoints/sentilo/cesva.py

@@ -0,0 +1,62 @@
+import logging
+import os
+from typing import Tuple, Union
+import json
+
+from .. import AsyncRequestHandler
+
+
+class RequestHandler(AsyncRequestHandler):
+    async def validate(
+        self, request_data: dict, endpoint_data: dict
+    ) -> Tuple[bool, Union[str, None], Union[int, None]]:
+        """
+        Use Starlette request_data here to determine should we accept or reject
+        this request
+        :param request_data: deserialized (FastAPI) Starlette Request
+        :param endpoint_data: endpoint data from device registry
+        :return: (bool ok, str error text, int status code)
+        """
+        [status_ok, response_message, status_code] = await super().validate(request_data, endpoint_data)
+
+        if status_ok is False:
+            return False, response_message, status_code
+
+        try:
+            # check if device id can be extracted
+            json.loads(request_data["request"]["body"].decode("utf-8"))["sensors"][0]["sensor"][0:-2]
+            return True, "Request accepted", 202
+        except Exception:
+            logging.warning("unable to retreive device_id from request body")
+            return False, "Invalid request, see logs for error", 400
+
+    async def process_request(
+        self,
+        request_data: dict,
+        endpoint_data: dict,
+    ) -> Tuple[bool, str, Union[str, None], Union[str, dict, list], int]:
+        auth_ok, response_message, status_code = await self.validate(request_data, endpoint_data)
+
+        logging.info("Validation: {}, {}, {}".format(auth_ok, response_message, status_code))
+        if auth_ok:
+            device_id = json.loads(request_data["request"]["body"].decode("utf-8"))["sensors"][0]["sensor"][0:-2]
+            topic_name = endpoint_data["kafka_raw_data_topic"]
+        else:
+            device_id = None
+            topic_name = None
+        return auth_ok, device_id, topic_name, response_message, status_code
+
+    async def get_metadata(self, request_data: dict, device_id: str) -> str:
+        # TODO: put this function to BaseRequestHandler or remove from endpoint
+        # (and add to parser)
+        metadata = "{}"
+        redis_url = os.getenv("REDIS_URL")
+        if redis_url is None:
+            logging.info("No REDIS_URL defined, querying device metadata failed")
+            return metadata
+        if device_id is None:
+            logging.info("No device_id available, querying device metadata failed")
+            return metadata
+        if metadata is None:
+            return "{}"
+        return metadata

tests/test_api_cesva.py

@@ -0,0 +1,73 @@
+import logging
+import os
+import json
+import httpx
+
+logging.basicConfig(level=logging.INFO)
+
+API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost:8001")
+API_TOKEN = os.getenv("API_TOKEN", "abc123")
+
+
+# query params
+PARAMS = {"x-api-key": API_TOKEN}
+
+# Body
+PAYLOAD = {
+    "sensors": [
+        {"sensor": "TA120-T246187-N", "observations": [{"value": "61.2", "timestamp": "24/02/2022T17:45:15UTC"}]},
+        {"sensor": "TA120-T246187-O", "observations": [{"value": "false", "timestamp": "24/02/2022T17:45:15UTC"}]},
+        {"sensor": "TA120-T246187-U", "observations": [{"value": "false", "timestamp": "24/02/2022T17:45:15UTC"}]},
+        {"sensor": "TA120-T246187-M", "observations": [{"value": "77", "timestamp": "24/02/2022T17:45:15UTC"}]},
+        {
+            "sensor": "TA120-T246187-S",
+            "observations": [
+                {
+                    "value": "060.6,0,0;060.8,0,0;060.4,0,0;059.9,0,0;059.9,0,0;060.6,0,0; \
+                    060.7,0,0;060.4,0,0;059.9,0,0;059.9,0,0;060.2,0,0;060.4,0,0;",
+                    "timestamp": "24/02/2022T17:45:15UTC",
+                }
+            ],
+        },
+    ]
+}
+
+
+def test_service_up():
+    url = API_BASE_URL
+    resp = httpx.get(url)
+    assert resp.status_code == 200, "service is up"
+    assert resp.json()["message"] == "Test ok", "service is up"
+
+
+def test_cesva_endpoint_up():
+    url = f"{API_BASE_URL}/api/v1/cesva"
+    resp = httpx.put(url)
+    assert resp.status_code == 401, "error: /api/v1/cesva accessible without token"
+    assert resp.text.startswith(
+        "Missing or invalid authentication token"
+    ), "error: /api/v1/cesva accessible without token"
+
+
+def test_cesva_endppoint_authenticated_access():
+    url = f"{API_BASE_URL}/api/v1/cesva"
+    payload = PAYLOAD.copy()
+    params = PARAMS.copy()
+
+    resp = httpx.put(url, params=params, data=json.dumps(payload))
+    logging.info(resp.text)
+    assert resp.status_code in [200, 201, 202], "message forwarded"
+    params["x-api-key"] = "wrong"
+    resp = httpx.put(url, params=params, data=payload)
+    logging.info(resp.text)
+    assert resp.status_code == 401, "failed as intended"
+
+
+def main():
+    # test_service_up()
+    # test_cesva_endpoint_up()
+    test_cesva_endppoint_authenticated_access()
+
+
+if __name__ == "__main__":
+    main()