From 149ec4e0dd1f7fbfe2a906e4edbc3ac63381d8fe Mon Sep 17 00:00:00 2001 From: DL6ER Date: Thu, 26 Jan 2023 20:11:55 +0100 Subject: [PATCH] Add test for re-importing the just exported Teleporter file during the tests Signed-off-by: DL6ER --- src/api/api.c | 5 ++- src/api/teleporter.c | 2 +- test/api/checkAPI.py | 20 ++++++++- test/api/libs/FTLAPI.py | 68 +++++++++++++++++++++++-------- test/api/libs/responseVerifyer.py | 36 ++++++++++++++-- test/run.sh | 4 +- 6 files changed, 107 insertions(+), 28 deletions(-) diff --git a/src/api/api.c b/src/api/api.c index 32c96afc..5356b756 100644 --- a/src/api/api.c +++ b/src/api/api.c @@ -100,10 +100,11 @@ int api_handler(struct mg_connection *conn, void *ignored) { false, false, 0 } }; - log_debug(DEBUG_API, "Requested API URI: %s %s ? %s", + log_debug(DEBUG_API, "Requested API URI: %s %s ? %s (Content-Type %s)", api.request->request_method, api.request->local_uri_raw, - api.request->query_string); + api.request->query_string, + mg_get_header(conn, "Content-Type")); int ret = 0; diff --git a/src/api/teleporter.c b/src/api/teleporter.c index cc177c67..6f7b6e5b 100644 --- a/src/api/teleporter.c +++ b/src/api/teleporter.c @@ -83,7 +83,7 @@ static int field_found(const char *key, is_file = true; return MG_FORM_FIELD_STORAGE_GET; } - else if(strcasecmp(key, "import") == 0) + else if(strcasecmp(key, "sid") == 0) { is_sid = true; return MG_FORM_FIELD_STORAGE_GET; diff --git a/test/api/checkAPI.py b/test/api/checkAPI.py index b081e6d4..bc628664 100644 --- a/test/api/checkAPI.py +++ b/test/api/checkAPI.py @@ -49,6 +49,7 @@ if __name__ == "__main__": # and have the same response format. Also verify that the examples # matches the OpenAPI specs. print("Verifying the individual endpoint properties...") + teleporter = None for path in openapi.endpoints["get"]: # We do not check the action endpoints as they'd trigger # possibly unwanted action such as restarting FTL, running @@ -57,15 +58,30 @@ if __name__ == "__main__": continue verifyer = ResponseVerifyer(ftl, openapi) errors = verifyer.verify_endpoint(path) + if verifyer.teleporter_archive is not None: + teleporter = verifyer.teleporter_archive if len(errors) == 0: - print(" " + path + " (" + verifyer.auth_method + " auth): OK") + print(" GET " + path + " (" + verifyer.auth_method + " auth): OK") else: - print(" " + path + " (" + verifyer.auth_method + " auth):") + print(" GET " + path + " (" + verifyer.auth_method + " auth):") for error in errors: print(" - " + error) errs[2] += len(errors) print("") + # Verify FTL Teleporter import + print("Verifying FTL Teleporter import...") + verifyer = ResponseVerifyer(ftl, openapi) + errors = verifyer.verify_teleporter_zip(teleporter) + if len(errors) == 0: + print(" POST /api/teleporter: OK") + else: + print(" Errors:") + for error in errors: + print(" - " + error) + errs[2] += len(errors) + + # Print the number error (if any) if errs[0] > 0: print("Found " + str(errs[0]) + " non-implemented endpoints") diff --git a/test/api/libs/FTLAPI.py b/test/api/libs/FTLAPI.py index d603c668..4a325a4a 100644 --- a/test/api/libs/FTLAPI.py +++ b/test/api/libs/FTLAPI.py @@ -10,6 +10,7 @@ # Please see LICENSE file for your rights under this license. from enum import Enum +import random import requests from typing import List import json @@ -27,12 +28,16 @@ sid = session["session"]["sid"] # SID string if succesful, null otherwise """ class AuthenticationMethods(Enum): + RANDOM = 0 HEADER = 1 BODY = 2 COOKIE = 3 # Class to query the FTL API class FTLAPI(): + + auth_method = "?" + def __init__(self, api_url: str): self.api_url = api_url self.endpoints = { @@ -90,7 +95,10 @@ class FTLAPI(): # Generate password hash pwhash = sha256(password.encode("ascii")).hexdigest() pwhash = sha256(pwhash.encode("ascii")).hexdigest() - print("Using password hash: \"" + pwhash + "\"") + print("Using password hash: " + pwhash) + + if len(pwhash) != 64: + raise Exception("Invalid length of password hash") # Get the challenge from FTL challenge = response["challenge"].encode("ascii") @@ -100,6 +108,31 @@ class FTLAPI(): raise Exception("FTL returned invalid challenge item") self.session = response["session"] + + def get_jsondata_headers_cookies(self, authenticate: AuthenticationMethods): + # Add session ID to the request (if any) + json_data = None + headers = None + cookies = None + if self.session is not None and 'sid' in self.session: + # Pick a random authentication method if requested + # Try again if the method comes out as random again + while authenticate == AuthenticationMethods.RANDOM: + authenticate = random.choice(list(AuthenticationMethods)) + + # Add the session ID to the request + if authenticate == AuthenticationMethods.HEADER: + headers = {"X-FTL-SID": self.session['sid']} + elif authenticate == AuthenticationMethods.BODY: + json_data = {"sid": self.session['sid'] } + elif authenticate == AuthenticationMethods.COOKIE: + cookies = {"sid": self.session['sid'] } + + self.auth_method = authenticate.name + + return json_data, headers, cookies + + # Query the FTL API (GET) and return the response def GET(self, uri: str, params: List[str] = [], expected_mimetype: str = "application/json", authenticate: AuthenticationMethods = AuthenticationMethods.BODY): self.errors = [] @@ -108,22 +141,14 @@ class FTLAPI(): if len(params) > 0: uri = uri + "?" + "&".join(params) - # Add session ID to the request (if any) - data = None - headers = None - cookies = None - if self.session is not None and 'sid' in self.session: - if authenticate == AuthenticationMethods.HEADER: - headers = {"X-FTL-SID": self.session['sid']} - elif authenticate == AuthenticationMethods.BODY: - data = {"sid": self.session['sid'] } - elif authenticate == AuthenticationMethods.COOKIE: - cookies = {"sid": self.session['sid'] } + # Get json_data, headers and cookies + json_data, headers, cookies = self.get_jsondata_headers_cookies(authenticate) + + if self.verbose: + print("GET " + self.api_url + uri + " with json_data: " + json.dumps(json_data)) # Query the API - if self.verbose: - print("GET " + self.api_url + uri + " with data: " + json.dumps(data)) - with requests.get(url = self.api_url + uri, json = data, headers=headers, cookies=cookies) as response: + with requests.get(url = self.api_url + uri, json = json_data, headers=headers, cookies=cookies) as response: if self.verbose: print(json.dumps(response.json(), indent=4)) if expected_mimetype == "application/json": @@ -134,13 +159,19 @@ class FTLAPI(): self.errors.append("Exception when GETing from FTL: " + str(e)) return None + # Query the FTL API (POST) and return the response - def POST(self, uri: str, data: dict = {}): + def POST(self, uri: str, json_data: dict = {}, authenticate: AuthenticationMethods = AuthenticationMethods.HEADER, files = None): self.errors = [] try: + # Get json_data, headers and cookies + _, headers, cookies = self.get_jsondata_headers_cookies(authenticate) + if self.verbose: - print("POST " + self.api_url + uri + " with data: " + json.dumps(data)) - with requests.post(url = self.api_url + uri, json = data) as response: + print("POST " + self.api_url + uri + " with json_data: " + json.dumps(json_data)) + + # Query the API + with requests.post(url = self.api_url + uri, json = json_data, files = files, headers=headers, cookies=cookies) as response: if self.verbose: print(json.dumps(response.json(), indent=4)) return response.json() @@ -148,6 +179,7 @@ class FTLAPI(): self.errors.append("Exception when POSTing to FTL: " + str(e)) return None + # Query the endpoints from FTL for comparison with the OpenAPI specs def get_endpoints(self): try: diff --git a/test/api/libs/responseVerifyer.py b/test/api/libs/responseVerifyer.py index e8d2a49b..da6b775e 100644 --- a/test/api/libs/responseVerifyer.py +++ b/test/api/libs/responseVerifyer.py @@ -10,6 +10,7 @@ # Please see LICENSE file for your rights under this license. import io +import pprint import random import zipfile from libs.openAPI import openApi @@ -21,9 +22,11 @@ class ResponseVerifyer(): # Translate between OpenAPI and Python types YAML_TYPES = { "string": [str], "integer": [int], "number": [int, float], "boolean": [bool], "array": [list] } - TELEPORTER_FILES = ["etc/pihole/gravity.db", "etc/pihole/pihole.toml", "etc/pihole/pihole-FTL.db", "etc/hosts"] + TELEPORTER_FILES_EXPORT = ["etc/pihole/gravity.db", "etc/pihole/pihole.toml", "etc/pihole/pihole-FTL.db", "etc/hosts"] + TELEPORTER_FILES_IMPORT = ['etc/pihole/pihole.toml', 'etc/pihole/dhcp.leases', 'etc/pihole/gravity.db'] auth_method = "?" + teleporter_archive = None def __init__(self, ftl: FTLAPI, openapi: openApi): self.ftl = ftl @@ -67,7 +70,6 @@ class ResponseVerifyer(): expected_mimetype = True # Assign random authentication method so we can test them all authentication_method = random.choice([a for a in AuthenticationMethods]) - self.auth_method = authentication_method.name # Check if the expected response is defined in the API specs response_rcode = self.openapi.paths[endpoint][method]['responses'][str(rcode)] if 'content' in response_rcode: @@ -103,6 +105,7 @@ class ResponseVerifyer(): # Get FTL response FTLresponse = self.ftl.GET("/api" + endpoint, FTLparameters, expected_mimetype, authentication_method) + self.auth_method = self.ftl.auth_method if FTLresponse is None: return self.ftl.errors @@ -152,7 +155,7 @@ class ResponseVerifyer(): # header block try: # Check if all expected files are present - for expected_file in self.TELEPORTER_FILES: + for expected_file in self.TELEPORTER_FILES_EXPORT: if expected_file not in zipfile_obj.namelist(): self.errors.append("File " + expected_file + " is missing in received archive.") pihole_toml = zipfile_obj.read("etc/pihole/pihole.toml") @@ -160,6 +163,9 @@ class ResponseVerifyer(): self.errors.append("Received ZIP file's pihole.toml starts with wrong header") except Exception as err: self.errors.append("Error during ZIP analysis: " + str(err)) + + # Store Teleporter archive for later use + self.teleporter_archive = FTLresponse else: self.errors.append("Checker script does not know how to check for mimetype \"" + expected_mimetype + "\"") @@ -167,6 +173,30 @@ class ResponseVerifyer(): return self.errors + def verify_teleporter_zip(self, teleporter_archive: bytes): + # Send the zip file to the FTL API + if teleporter_archive is None: + self.errors.append("No Teleporter archive available for verification") + return self.errors + + # Send the archive to the FTL API + FTLresponse = self.ftl.POST("/api/teleporter", None, AuthenticationMethods.HEADER, {"file": ('teleporter.zip', teleporter_archive, 'application/zip')}) + + #Compare the response with the expected response + if FTLresponse is None: + self.errors.append("No response from FTL API") + return self.errors + if 'files' not in FTLresponse: + self.errors.append("Missing 'files' key in FTL response") + return self.errors + # Compare FTLresponse['files'] with self.TELEPORTER_FILES_IMPORT + for expected_file in self.TELEPORTER_FILES_IMPORT: + if expected_file not in FTLresponse['files']: + self.errors.append("File " + expected_file + " is missing in FTL response") + + return self.errors + + # Verify a single property's type def verify_type(self, prop_type: any, yaml_type: str, yaml_nullable: bool): # None is an acceptable reply when this is specified in the API specs diff --git a/test/run.sh b/test/run.sh index f7bb81bf..9418ae87 100755 --- a/test/run.sh +++ b/test/run.sh @@ -27,9 +27,9 @@ mkdir -p /home/pihole /etc/pihole /run/pihole /var/log/pihole echo "" > /var/log/pihole/FTL.log echo "" > /var/log/pihole/pihole.log touch /run/pihole-FTL.pid /run/pihole-FTL.port dig.log ptr.log -touch /var/log/pihole/HTTP_info.log /var/log/pihole/PH7.log +touch /var/log/pihole/HTTP_info.log /var/log/pihole/PH7.log /etc/pihole/dhcp.leases chown pihole:pihole /etc/pihole /run/pihole /var/log/pihole/pihole.log /var/log/pihole/FTL.log /run/pihole-FTL.pid /run/pihole-FTL.port -chown pihole:pihole /var/log/pihole/HTTP_info.log /var/log/pihole/PH7.log +chown pihole:pihole /var/log/pihole/HTTP_info.log /var/log/pihole/PH7.log /etc/pihole/dhcp.leases # Copy binary into a location the new user pihole can access cp ./pihole-FTL /home/pihole/pihole-FTL