From 66b366e1d64f8d9f7c4d16b35102ac4a425487a4 Mon Sep 17 00:00:00 2001 From: David Wharton Date: Sat, 14 Nov 2026 07:52:00 +0000 Subject: [PATCH 1/6] Fix API client retries and return the job ID Every request ran RETRIES times, even on success, so submit_job queued three jobs; retryable errors slept and then raised without retrying. Requests are now made once, retried on 429/5xx (job submissions only on 429/503, so a job can't be queued twice), rewind uploaded files before a retry, and time out. submit_job returns the job ID and raises DaltonAPIError on Dalton's error page. --- README.rst | 12 +++- api/dalton.py | 97 +++++++++++++++++--------- api/examples/job_submission.py | 3 +- pyproject.toml | 2 + tests/test_api_client.py | 124 +++++++++++++++++++++++++++++++++ 5 files changed, 203 insertions(+), 35 deletions(-) create mode 100644 tests/test_api_client.py diff --git a/README.rst b/README.rst index 25e76ca..233826c 100644 --- a/README.rst +++ b/README.rst @@ -896,7 +896,15 @@ Dalton API Client ----------------- An API Client has been added in ``api/dalton.py`` that performs API calls with Python requests. -The client is limited to GET and POST requests. +The client is limited to GET and POST requests. It needs the ``requests`` package. + +``submit_job()`` submits a teapot job and returns its job ID (several, comma-separated, if Dalton +split the pcap). If Dalton returns an error page instead, it raises ``DaltonAPIError`` with the +page's text. + +Requests time out after 60 seconds. A request that gets a 429 or 5xx response is retried up to +three times, 60 seconds apart. Job submissions are retried only on 429 and 503, which mean the +job was not queued, so a retry can't submit the same job twice. Submit Job using the API ------------------------ @@ -934,7 +942,7 @@ Other useful arguments to submit a job are: - ``custom_rules`` in which you may include the custom rules you may want to test with your job, - ``optionAlertDetailed``, ``optionEveLog``, ``optionOtherLogs``: this can be set to ``True`` if you want to generate additional logs with your job. -An example script can be found in ``api/examples/submit_job.py``. +An example script can be found in ``api/examples/job_submission.py``. Teapot Jobs =========== diff --git a/api/dalton.py b/api/dalton.py index 42b254c..32ea10e 100644 --- a/api/dalton.py +++ b/api/dalton.py @@ -1,5 +1,6 @@ """Dalton API client.""" +import re import time import requests @@ -7,47 +8,58 @@ RETRIES = 3 SLEEP_TIME = 60 +TIMEOUT = 60 +# GETs are retried on any of these. A job submission (POST) is retried only on +# the codes that mean the request wasn't processed, so a retry can't create a +# second job. +RETRY_CODES_GET = (429, 500, 502, 503, 504) +RETRY_CODES_POST = (429, 503) -class DaltonAPI: - def __init__(self, api_link): - self.api_link = api_link +# what a teapot job submission returns: a job ID, or several, comma-separated, +# when the pcap was split +TEAPOT_JIDS_RE = re.compile(r"^teapot_[0-9a-f]{16}(,teapot_[0-9a-f]{16})*$") - def _dalton_get(self, endpoint: str) -> requests.Response: - for _ in range(RETRIES): - try: - response = requests.get(url=f"{self.api_link}/{endpoint}") - response.raise_for_status() - except HTTPError as exc: - code = exc.response.status_code - if code in [429, 500, 502, 503, 504]: - # retry after n seconds - time.sleep(SLEEP_TIME) +class DaltonAPIError(Exception): + """Dalton rejected a request without an HTTP error status.""" - raise - - return response - def _dalton_post(self, endpoint: str, data: dict, files: dict) -> requests.Response: - for _ in range(RETRIES): +class DaltonAPI: + def __init__(self, api_link, timeout=TIMEOUT): + self.api_link = api_link + self.timeout = timeout + + def _request( + self, method: str, endpoint: str, retry_codes, **kwargs + ) -> requests.Response: + """Make a request, retrying after SLEEP_TIME seconds on a retryable status.""" + for attempt in range(1, RETRIES + 1): + # a retried upload has to send its files from the start again + for value in (kwargs.get("files") or {}).values(): + fileobj = value[1] if isinstance(value, tuple) else value + if hasattr(fileobj, "seek"): + fileobj.seek(0) + response = requests.request( + method, + url=f"{self.api_link}/{endpoint}", + timeout=self.timeout, + **kwargs, + ) try: - response = requests.post( - url=f"{self.api_link}/{endpoint}", - data=data, - files=files, - ) response.raise_for_status() - except HTTPError as exc: - code = exc.response.status_code - - if code in [429, 500, 502, 503, 504]: - # retry after n seconds + except HTTPError: + if response.status_code in retry_codes and attempt < RETRIES: time.sleep(SLEEP_TIME) - + continue raise + return response - return response + def _dalton_get(self, endpoint: str) -> requests.Response: + return self._request("GET", endpoint, RETRY_CODES_GET) + + def _dalton_post(self, endpoint: str, data: dict, files: dict) -> requests.Response: + return self._request("POST", endpoint, RETRY_CODES_POST, data=data, files=files) def submit_job( self, @@ -56,7 +68,7 @@ def submit_job( configuration: str, custom_rules: str, files: dict, - ) -> None: + ) -> str: """Submits a job for a specific sensor tech, ruleset, configuration and pcap. Args: @@ -65,6 +77,13 @@ def submit_job( configuration (str): sensor configuration. Can be read from a yaml file. See "examples/job_submission.py" custom_rules (str): rules to be tested. Can be read from a rules file. See "examples/job_submission.py" files (dict): dictionary that includes tuple with pcap file. Ex. {"coverage-pcap0": ("test_job.pcap", pcap)} + + Returns: + str: the job ID, or a comma-separated list of job IDs if Dalton split the pcap. + Ex. 'teapot_0123456789abcdef' + + Raises: + DaltonAPIError: Dalton returned an error page instead of a job ID. """ data = { "sensor_tech": sensor_tech, @@ -74,7 +93,14 @@ def submit_job( "custom_ruleset": custom_rules, "teapotJob": 1, } - self._dalton_post("dalton/coverage/summary", data=data, files=files) + response = self._dalton_post("dalton/coverage/summary", data=data, files=files) + # a teapot job's response is its job ID; anything else is Dalton's error page + jids = response.text.strip() + if not TEAPOT_JIDS_RE.match(jids): + raise DaltonAPIError( + f"Dalton did not accept the job: {_page_text(jids)[:500]}" + ) + return jids def get_sensor_tech(self, sensor: str) -> list: """Get sensor type and version. @@ -146,3 +172,10 @@ def get_current_sensors(self) -> dict: "dalton/controller_api/get-current-sensors-json-full" ) return response.json() + + +def _page_text(html: str) -> str: + """The visible text of an HTML page, whitespace collapsed.""" + text = re.sub(r"(?is)<(script|style)\b.*?", " ", html) + text = re.sub(r"<[^>]+>", " ", text) + return " ".join(text.split()) diff --git a/api/examples/job_submission.py b/api/examples/job_submission.py index ec9c448..ac212ba 100644 --- a/api/examples/job_submission.py +++ b/api/examples/job_submission.py @@ -29,10 +29,11 @@ # have to reread the file to send it to multiple instances of the sensor pcap = open(f"{CURRENT_PATH}/api/examples/mocks/test_job.pcap", "rb") files = {"coverage-pcap0": ("test_job.pcap", pcap)} - dalton_client.submit_job( + jid = dalton_client.submit_job( sensor_tech=tech, prod_ruleset=dalton_rulesets[0], configuration=configuration, custom_rules=custom_rules, files=files, ) + print(f"{tech}: {jid}") diff --git a/pyproject.toml b/pyproject.toml index e628b43..e9fab01 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,8 @@ Homepage = "https://github.com/secureworks/dalton" [project.optional-dependencies] testing = [ + # api/dalton.py, the API client + "requests", "pytest", ] devtools = [ diff --git a/tests/test_api_client.py b/tests/test_api_client.py new file mode 100644 index 0000000..72e73ff --- /dev/null +++ b/tests/test_api_client.py @@ -0,0 +1,124 @@ +"""Tests for the Dalton API client in api/dalton.py.""" + +import io + +import pytest +import requests + +from api import dalton +from api.dalton import DaltonAPI, DaltonAPIError + +JID = "teapot_0123456789abcdef" + + +def make_response(status=200, text=""): + response = requests.Response() + response.status_code = status + response._content = text.encode() + response.url = "http://dalton.test/" + return response + + +class FakeRequests: + """Stands in for requests.request: replays responses and records calls.""" + + def __init__(self, *responses): + self.responses = list(responses) + self.calls = [] + + def __call__(self, method, url, **kwargs): + files = kwargs.get("files") or {} + sent = {name: value[1].read() for name, value in files.items()} + self.calls.append({"method": method, "url": url, **kwargs, "sent": sent}) + return self.responses.pop(0) + + +@pytest.fixture +def fake(monkeypatch): + def install(*responses): + fake_requests = FakeRequests(*responses) + monkeypatch.setattr(dalton.requests, "request", fake_requests) + return fake_requests + + monkeypatch.setattr(dalton.time, "sleep", lambda seconds: None) + return install + + +def submit(client, pcap=b"pcap bytes"): + return client.submit_job( + sensor_tech="suricata/8.0.6", + prod_ruleset="/opt/dalton/rulesets/suricata/x.rules", + configuration="%YAML 1.1\n", + custom_rules='alert ip any any -> any any (msg:"x"; sid:1;)', + files={"coverage-pcap0": ("test.pcap", io.BytesIO(pcap))}, + ) + + +def test_get_is_made_once(fake): + requests_ = fake(make_response(text='{"sensor_tech": ["suricata/8.0.6"]}')) + assert DaltonAPI("http://dalton.test").get_sensor_tech("suricata") == [ + "suricata/8.0.6" + ] + assert len(requests_.calls) == 1 + call = requests_.calls[0] + assert (call["method"], call["url"]) == ( + "GET", + "http://dalton.test/dalton/controller_api/get-current-sensors/suricata", + ) + assert call["timeout"] == dalton.TIMEOUT + + +def test_submit_returns_job_id_and_posts_once(fake): + requests_ = fake(make_response(text=JID)) + assert submit(DaltonAPI("http://dalton.test")) == JID + assert len(requests_.calls) == 1 + assert requests_.calls[0]["data"]["teapotJob"] == 1 + + +def test_submit_split_pcap_job_ids(fake): + jids = f"{JID},teapot_fedcba9876543210" + fake(make_response(text=jids)) + assert submit(DaltonAPI("http://dalton.test")) == jids + + +def test_submit_error_page(fake): + page = "

Error: no sensor

" + fake(make_response(text=page)) + with pytest.raises(DaltonAPIError, match="Error: no sensor"): + submit(DaltonAPI("http://dalton.test")) + + +def test_get_retries_then_succeeds(fake): + requests_ = fake(make_response(500), make_response(503), make_response(text="5")) + assert DaltonAPI("http://dalton.test").get_max_pcaps() == 5 + assert len(requests_.calls) == 3 + + +def test_get_gives_up_after_retries(fake): + requests_ = fake(*[make_response(502) for _ in range(dalton.RETRIES)]) + with pytest.raises(requests.HTTPError): + DaltonAPI("http://dalton.test").get_max_pcaps() + assert len(requests_.calls) == dalton.RETRIES + + +def test_not_found_is_not_retried(fake): + requests_ = fake(make_response(404)) + with pytest.raises(requests.HTTPError): + DaltonAPI("http://dalton.test").get_max_pcaps() + assert len(requests_.calls) == 1 + + +def test_submit_retry_resends_the_whole_pcap(fake): + requests_ = fake(make_response(503), make_response(text=JID)) + assert submit(DaltonAPI("http://dalton.test"), pcap=b"0123456789") == JID + assert [call["sent"]["coverage-pcap0"] for call in requests_.calls] == [ + b"0123456789" + ] * 2 + + +def test_submit_not_retried_when_it_may_have_run(fake): + # a 500 may come after the job was queued; retrying could submit it twice + requests_ = fake(make_response(500)) + with pytest.raises(requests.HTTPError): + submit(DaltonAPI("http://dalton.test")) + assert len(requests_.calls) == 1 From c7170c6d81dda08892eb9c5d769f7db362829bc5 Mon Sep 17 00:00:00 2001 From: David Wharton Date: Sat, 14 Nov 2026 13:52:54 +0000 Subject: [PATCH 2/6] API client: no submission read timeout, retry connection failures, send custom rules - job submissions wait for Dalton's reply by default (submit_timeout sets a limit) - GETs are retried on connection errors and timeouts; submissions only when the connection failed before anything was sent - honour Retry-After; accept files as a list; reuse one Session - set optionCustomRuleset, without which Dalton ignored the custom rules - the example takes the pcap as an argument (its mock pcap was never committed) --- README.rst | 18 +++-- api/dalton.py | 98 ++++++++++++++++------ api/examples/job_submission.py | 42 ++++++---- tests/test_api_client.py | 143 ++++++++++++++++++++++++++++++--- 4 files changed, 246 insertions(+), 55 deletions(-) diff --git a/README.rst b/README.rst index 233826c..2b172bb 100644 --- a/README.rst +++ b/README.rst @@ -902,9 +902,15 @@ The client is limited to GET and POST requests. It needs the ``requests`` packag split the pcap). If Dalton returns an error page instead, it raises ``DaltonAPIError`` with the page's text. -Requests time out after 60 seconds. A request that gets a 429 or 5xx response is retried up to -three times, 60 seconds apart. Job submissions are retried only on 429 and 503, which mean the -job was not queued, so a retry can't submit the same job twice. +Each request is tried up to three times. A GET is retried after a 429 or 5xx response, a +connection error or a timeout. A job submission is retried only when the request can't have +queued the job: a 429 or 503 response, or a failure to connect. Retries wait for the server's +``Retry-After``, if it gives one, or 60 seconds. + +The client waits up to 60 seconds to connect and for the reply to a GET (``DaltonAPI(url, +timeout=...)``). Dalton processes the pcap before it replies to a job submission, so by default +there's no limit on that wait; ``submit_timeout=...`` sets one, but a submission that times out +may still have been queued. Submit Job using the API ------------------------ @@ -939,10 +945,12 @@ Submit the job as a shortlived ``teapotJob`` if you plan to make multiple calls Other useful arguments to submit a job are: -- ``custom_rules`` in which you may include the custom rules you may want to test with your job, +- ``custom_ruleset`` in which you may include the custom rules you may want to test with your job, + along with ``optionCustomRuleset`` set to ``custom`` (without it, the custom rules are ignored), - ``optionAlertDetailed``, ``optionEveLog``, ``optionOtherLogs``: this can be set to ``True`` if you want to generate additional logs with your job. -An example script can be found in ``api/examples/job_submission.py``. +An example script can be found in ``api/examples/job_submission.py``. Run it from the repository +root with a pcap: ``python -m api.examples.job_submission ``. Teapot Jobs =========== diff --git a/api/dalton.py b/api/dalton.py index 32ea10e..63d9ac1 100644 --- a/api/dalton.py +++ b/api/dalton.py @@ -4,7 +4,7 @@ import time import requests -from requests.exceptions import HTTPError +from urllib3.exceptions import ConnectTimeoutError RETRIES = 3 SLEEP_TIME = 60 @@ -26,40 +26,65 @@ class DaltonAPIError(Exception): class DaltonAPI: - def __init__(self, api_link, timeout=TIMEOUT): + def __init__(self, api_link, timeout=TIMEOUT, submit_timeout=None): + """ + Args: + api_link (str): Dalton base URL. Ex. http://localhost + timeout (float): seconds to wait for a connection, and for a reply to a GET + submit_timeout (float): seconds to wait for the reply to a job submission. + Dalton processes the pcap before it replies, so the default is no limit; + a submission that times out may still have been queued. + """ self.api_link = api_link self.timeout = timeout + self.submit_timeout = submit_timeout + self.session = requests.Session() def _request( - self, method: str, endpoint: str, retry_codes, **kwargs + self, method: str, endpoint: str, retry_codes, timeout, **kwargs ) -> requests.Response: - """Make a request, retrying after SLEEP_TIME seconds on a retryable status.""" + """Make a request, trying up to RETRIES times. + + A retryable status waits for the server's Retry-After, if it gives one in + seconds, or SLEEP_TIME. Connection failures are retried after SLEEP_TIME: + for a GET any failure, for a POST only one that happened before the + request was sent. + """ for attempt in range(1, RETRIES + 1): - # a retried upload has to send its files from the start again - for value in (kwargs.get("files") or {}).values(): - fileobj = value[1] if isinstance(value, tuple) else value - if hasattr(fileobj, "seek"): - fileobj.seek(0) - response = requests.request( - method, - url=f"{self.api_link}/{endpoint}", - timeout=self.timeout, - **kwargs, - ) + last = attempt == RETRIES + _rewind(kwargs.get("files")) try: - response.raise_for_status() - except HTTPError: - if response.status_code in retry_codes and attempt < RETRIES: - time.sleep(SLEEP_TIME) - continue + response = self.session.request( + method, + url=f"{self.api_link}/{endpoint}", + timeout=timeout, + **kwargs, + ) + except requests.exceptions.SSLError: raise + except (requests.ConnectionError, requests.Timeout) as exc: + if last or not (method == "GET" or _never_sent(exc)): + raise + time.sleep(SLEEP_TIME) + continue + if response.status_code in retry_codes and not last: + time.sleep(_retry_after(response)) + continue + response.raise_for_status() return response def _dalton_get(self, endpoint: str) -> requests.Response: - return self._request("GET", endpoint, RETRY_CODES_GET) + return self._request("GET", endpoint, RETRY_CODES_GET, self.timeout) def _dalton_post(self, endpoint: str, data: dict, files: dict) -> requests.Response: - return self._request("POST", endpoint, RETRY_CODES_POST, data=data, files=files) + return self._request( + "POST", + endpoint, + RETRY_CODES_POST, + (self.timeout, self.submit_timeout), + data=data, + files=files, + ) def submit_job( self, @@ -76,7 +101,8 @@ def submit_job( prod_ruleset (str): path to prod ruleset. Ex. /opt/dalton/rulesets/suricata/ET-20230301-all-suricata.rules configuration (str): sensor configuration. Can be read from a yaml file. See "examples/job_submission.py" custom_rules (str): rules to be tested. Can be read from a rules file. See "examples/job_submission.py" - files (dict): dictionary that includes tuple with pcap file. Ex. {"coverage-pcap0": ("test_job.pcap", pcap)} + files (dict): dictionary that includes tuple with pcap file. Ex. {"coverage-pcap0": ("test_job.pcap", pcap)}. + A list of (name, tuple) pairs also works. Open files are rewound before a retry. Returns: str: the job ID, or a comma-separated list of job IDs if Dalton split the pcap. @@ -93,6 +119,9 @@ def submit_job( "custom_ruleset": custom_rules, "teapotJob": 1, } + if custom_rules: + # Dalton ignores custom_ruleset unless this is set + data["optionCustomRuleset"] = "custom" response = self._dalton_post("dalton/coverage/summary", data=data, files=files) # a teapot job's response is its job ID; anything else is Dalton's error page jids = response.text.strip() @@ -179,3 +208,26 @@ def _page_text(html: str) -> str: text = re.sub(r"(?is)<(script|style)\b.*?", " ", html) text = re.sub(r"<[^>]+>", " ", text) return " ".join(text.split()) + + +def _rewind(files) -> None: + """Seek open files back to the start, so a retry sends them whole.""" + if not files: + return + values = files.values() if hasattr(files, "values") else (v for _, v in files) + for value in values: + fileobj = value[1] if isinstance(value, tuple) else value + if hasattr(fileobj, "seek"): + fileobj.seek(0) + + +def _never_sent(exc: Exception) -> bool: + """Whether a request failed while connecting, before the server saw it.""" + reason = getattr(exc.args[0], "reason", None) if exc.args else None + return isinstance(reason, ConnectTimeoutError) + + +def _retry_after(response: requests.Response) -> float: + """Seconds to wait before retrying: the server's Retry-After, or SLEEP_TIME.""" + value = response.headers.get("Retry-After", "").strip() + return int(value) if value.isdigit() else SLEEP_TIME diff --git a/api/examples/job_submission.py b/api/examples/job_submission.py index ac212ba..f3e5abb 100644 --- a/api/examples/job_submission.py +++ b/api/examples/job_submission.py @@ -1,6 +1,10 @@ -"""Example on how to submit a job using the Dalton Client API. Mock data is in mocks directory.""" +"""Example on how to submit a job using the Dalton Client API. Mock data is in mocks directory. + +Run it from the repository root: python -m api.examples.job_submission +""" import os +import sys from api.dalton import DaltonAPI @@ -8,16 +12,20 @@ SENSOR_NAME = "suricata" CURRENT_PATH = os.getcwd() +if len(sys.argv) != 2: + sys.exit("usage: python -m api.examples.job_submission ") +PCAP = sys.argv[1] + # create API client dalton_client = DaltonAPI(DALTON_URL) # this file needs to include the sensor configuration in yaml format. Make sure that the files are in the proper directory. -conf = open(f"{CURRENT_PATH}/api/examples/mocks/engine_conf.yaml", "r") -configuration = conf.read() +with open(f"{CURRENT_PATH}/api/examples/mocks/engine_conf.yaml", "r") as conf: + configuration = conf.read() # this file needs to include any custom rules you want to test -cust = open(f"{CURRENT_PATH}/api/examples/mocks/custom_rules.rules", "r") -custom_rules = cust.read() +with open(f"{CURRENT_PATH}/api/examples/mocks/custom_rules.rules", "r") as cust: + custom_rules = cust.read() # get rulesets and sensor techs to use in job submission parameters dalton_rulesets = dalton_client.get_prod_rulesets(SENSOR_NAME) @@ -25,15 +33,15 @@ # since techs and rulesets are a list we use a loop to get the corresponding tech and ruleset # you will probably use this if you want to test with multiple versions of the sensor -for tech in dalton_sensor_techs: - # have to reread the file to send it to multiple instances of the sensor - pcap = open(f"{CURRENT_PATH}/api/examples/mocks/test_job.pcap", "rb") - files = {"coverage-pcap0": ("test_job.pcap", pcap)} - jid = dalton_client.submit_job( - sensor_tech=tech, - prod_ruleset=dalton_rulesets[0], - configuration=configuration, - custom_rules=custom_rules, - files=files, - ) - print(f"{tech}: {jid}") +with open(PCAP, "rb") as pcap: + for tech in dalton_sensor_techs: + # the client rewinds the pcap before each upload, so one open file serves every job + files = {"coverage-pcap0": (os.path.basename(PCAP), pcap)} + jid = dalton_client.submit_job( + sensor_tech=tech, + prod_ruleset=dalton_rulesets[0], + configuration=configuration, + custom_rules=custom_rules, + files=files, + ) + print(f"{tech}: {jid}") diff --git a/tests/test_api_client.py b/tests/test_api_client.py index 72e73ff..2b7f5eb 100644 --- a/tests/test_api_client.py +++ b/tests/test_api_client.py @@ -1,9 +1,12 @@ """Tests for the Dalton API client in api/dalton.py.""" import io +import socket +import threading import pytest import requests +from urllib3.exceptions import MaxRetryError, NewConnectionError, ProtocolError from api import dalton from api.dalton import DaltonAPI, DaltonAPIError @@ -11,16 +14,27 @@ JID = "teapot_0123456789abcdef" -def make_response(status=200, text=""): +def make_response(status=200, text="", headers=None): response = requests.Response() response.status_code = status + response.headers.update(headers or {}) response._content = text.encode() response.url = "http://dalton.test/" return response +# a refused connection, which never reached the server +REFUSED = requests.ConnectionError( + MaxRetryError(None, "/", NewConnectionError(None, "Connection refused")) +) +# a connection dropped after the request was sent +DROPPED = requests.ConnectionError( + ProtocolError("Connection aborted.", ConnectionResetError()) +) + + class FakeRequests: - """Stands in for requests.request: replays responses and records calls.""" + """Stands in for Session.request: replays responses (or raises) and records calls.""" def __init__(self, *responses): self.responses = list(responses) @@ -28,29 +42,39 @@ def __init__(self, *responses): def __call__(self, method, url, **kwargs): files = kwargs.get("files") or {} - sent = {name: value[1].read() for name, value in files.items()} + pairs = files.items() if hasattr(files, "items") else files + sent = {name: value[1].read() for name, value in pairs} self.calls.append({"method": method, "url": url, **kwargs, "sent": sent}) - return self.responses.pop(0) + response = self.responses.pop(0) + if isinstance(response, Exception): + raise response + return response + + +@pytest.fixture +def sleeps(monkeypatch): + slept = [] + monkeypatch.setattr(dalton.time, "sleep", slept.append) + return slept @pytest.fixture -def fake(monkeypatch): +def fake(monkeypatch, sleeps): def install(*responses): fake_requests = FakeRequests(*responses) - monkeypatch.setattr(dalton.requests, "request", fake_requests) + monkeypatch.setattr(requests.Session, "request", fake_requests) return fake_requests - monkeypatch.setattr(dalton.time, "sleep", lambda seconds: None) return install -def submit(client, pcap=b"pcap bytes"): +def submit(client, pcap=b"pcap bytes", files=None): return client.submit_job( sensor_tech="suricata/8.0.6", prod_ruleset="/opt/dalton/rulesets/suricata/x.rules", configuration="%YAML 1.1\n", custom_rules='alert ip any any -> any any (msg:"x"; sid:1;)', - files={"coverage-pcap0": ("test.pcap", io.BytesIO(pcap))}, + files=files or {"coverage-pcap0": ("test.pcap", io.BytesIO(pcap))}, ) @@ -72,7 +96,18 @@ def test_submit_returns_job_id_and_posts_once(fake): requests_ = fake(make_response(text=JID)) assert submit(DaltonAPI("http://dalton.test")) == JID assert len(requests_.calls) == 1 - assert requests_.calls[0]["data"]["teapotJob"] == 1 + data = requests_.calls[0]["data"] + assert data["teapotJob"] == 1 + assert data["optionCustomRuleset"] == "custom" + assert data["custom_ruleset"].startswith("alert ") + # Dalton processes the pcap before replying: no read timeout by default + assert requests_.calls[0]["timeout"] == (dalton.TIMEOUT, None) + + +def test_submit_timeout_is_configurable(fake): + requests_ = fake(make_response(text=JID)) + submit(DaltonAPI("http://dalton.test", timeout=5, submit_timeout=600)) + assert requests_.calls[0]["timeout"] == (5, 600) def test_submit_split_pcap_job_ids(fake): @@ -122,3 +157,91 @@ def test_submit_not_retried_when_it_may_have_run(fake): with pytest.raises(requests.HTTPError): submit(DaltonAPI("http://dalton.test")) assert len(requests_.calls) == 1 + + +def test_submit_files_as_a_list(fake): + pcap = io.BytesIO(b"0123456789") + requests_ = fake(make_response(503), make_response(text=JID)) + files = [("coverage-pcap0", ("test.pcap", pcap))] + assert submit(DaltonAPI("http://dalton.test"), files=files) == JID + assert [call["sent"]["coverage-pcap0"] for call in requests_.calls] == [ + b"0123456789" + ] * 2 + + +def test_get_retries_connection_errors(fake): + requests_ = fake(DROPPED, requests.ReadTimeout(), make_response(text="5")) + assert DaltonAPI("http://dalton.test").get_max_pcaps() == 5 + assert len(requests_.calls) == 3 + + +def test_submit_retried_when_connection_refused(fake): + requests_ = fake(REFUSED, make_response(text=JID)) + assert submit(DaltonAPI("http://dalton.test")) == JID + assert len(requests_.calls) == 2 + + +def test_submit_not_retried_when_connection_dropped(fake): + # the request was sent, so the job may have been queued + requests_ = fake(DROPPED) + with pytest.raises(requests.ConnectionError): + submit(DaltonAPI("http://dalton.test")) + assert len(requests_.calls) == 1 + + +def test_tls_errors_are_not_retried(fake): + requests_ = fake(requests.exceptions.SSLError("certificate verify failed")) + with pytest.raises(requests.exceptions.SSLError): + DaltonAPI("https://dalton.test").get_max_pcaps() + assert len(requests_.calls) == 1 + + +def test_retry_after_is_honoured(fake, sleeps): + fake( + make_response(429, headers={"Retry-After": "7"}), + make_response(503, headers={"Retry-After": "Wed, 21 Oct 2026 07:28:00 GMT"}), + make_response(text="5"), + ) + assert DaltonAPI("http://dalton.test").get_max_pcaps() == 5 + assert sleeps == [7, dalton.SLEEP_TIME] + + +def test_never_sent_with_real_connections(): + # a port nothing listens on refuses the connection + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + closed_port = probe.getsockname()[1] + with pytest.raises(requests.ConnectionError) as refused: + requests.post(f"http://127.0.0.1:{closed_port}/", data={"a": 1}, timeout=5) + assert dalton._never_sent(refused.value) + + # a server that reads the request and hangs up without replying + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.listen() + + def hang_up(): + conn, _ = server.accept() + conn.recv(65536) + conn.close() + + thread = threading.Thread(target=hang_up, daemon=True) + thread.start() + port = server.getsockname()[1] + with pytest.raises(requests.ConnectionError) as dropped: + requests.post(f"http://127.0.0.1:{port}/", data={"a": 1}, timeout=5) + thread.join() + server.close() + assert not dalton._never_sent(dropped.value) + + +def test_submit_without_custom_rules(fake): + requests_ = fake(make_response(text=JID)) + DaltonAPI("http://dalton.test").submit_job( + sensor_tech="suricata/8.0.6", + prod_ruleset="/opt/dalton/rulesets/suricata/x.rules", + configuration="%YAML 1.1\n", + custom_rules="", + files={"coverage-pcap0": ("test.pcap", io.BytesIO(b"x"))}, + ) + assert "optionCustomRuleset" not in requests_.calls[0]["data"] From e5f6f1f68b65a7eb984bc64a0b98b7a10ee85fce Mon Sep 17 00:00:00 2001 From: David Wharton Date: Sat, 14 Nov 2026 14:56:47 +0000 Subject: [PATCH 3/6] API client: fix second-review findings - fresh connection per request (no Session), with a headers= option - read uploads once from their current position instead of rewinding - connect timeout taken from a (connect, read) timeout tuple - retry proxy connection failures; cap Retry-After at 300s - error messages show Dalton's message only; RETRIES=0 still makes one attempt --- README.rst | 7 ++- api/dalton.py | 81 ++++++++++++++---------- api/examples/job_submission.py | 23 +++---- tests/test_api_client.py | 109 +++++++++++++++++++++++++++++---- 4 files changed, 162 insertions(+), 58 deletions(-) diff --git a/README.rst b/README.rst index 2b172bb..cc59803 100644 --- a/README.rst +++ b/README.rst @@ -900,17 +900,18 @@ The client is limited to GET and POST requests. It needs the ``requests`` packag ``submit_job()`` submits a teapot job and returns its job ID (several, comma-separated, if Dalton split the pcap). If Dalton returns an error page instead, it raises ``DaltonAPIError`` with the -page's text. +error message. Pcaps can be passed as bytes or open files; a file is read from its current position, +so pass bytes to submit the same pcap more than once. Each request is tried up to three times. A GET is retried after a 429 or 5xx response, a connection error or a timeout. A job submission is retried only when the request can't have queued the job: a 429 or 503 response, or a failure to connect. Retries wait for the server's -``Retry-After``, if it gives one, or 60 seconds. +``Retry-After``, if it gives one in seconds (at most 300), or 60 seconds. The client waits up to 60 seconds to connect and for the reply to a GET (``DaltonAPI(url, timeout=...)``). Dalton processes the pcap before it replies to a job submission, so by default there's no limit on that wait; ``submit_timeout=...`` sets one, but a submission that times out -may still have been queued. +may still have been queued. ``headers={...}`` adds headers, such as an API key, to every request. Submit Job using the API ------------------------ diff --git a/api/dalton.py b/api/dalton.py index 63d9ac1..5d769c7 100644 --- a/api/dalton.py +++ b/api/dalton.py @@ -1,14 +1,17 @@ """Dalton API client.""" +import html import re import time import requests -from urllib3.exceptions import ConnectTimeoutError +from urllib3.exceptions import ConnectTimeoutError, ProxyError RETRIES = 3 SLEEP_TIME = 60 TIMEOUT = 60 +# longest wait a server's Retry-After can ask for +MAX_RETRY_AFTER = 300 # GETs are retried on any of these. A job submission (POST) is retried only on # the codes that mean the request wasn't processed, so a retry can't create a @@ -26,19 +29,21 @@ class DaltonAPIError(Exception): class DaltonAPI: - def __init__(self, api_link, timeout=TIMEOUT, submit_timeout=None): + def __init__(self, api_link, timeout=TIMEOUT, submit_timeout=None, headers=None): """ Args: api_link (str): Dalton base URL. Ex. http://localhost - timeout (float): seconds to wait for a connection, and for a reply to a GET + timeout (float): seconds to wait for a connection, and for a reply to a GET. + A (connect, read) tuple sets the two separately. submit_timeout (float): seconds to wait for the reply to a job submission. Dalton processes the pcap before it replies, so the default is no limit; a submission that times out may still have been queued. + headers (dict): extra headers sent with every request, e.g. an API key """ self.api_link = api_link self.timeout = timeout self.submit_timeout = submit_timeout - self.session = requests.Session() + self.headers = headers def _request( self, method: str, endpoint: str, retry_codes, timeout, **kwargs @@ -46,17 +51,19 @@ def _request( """Make a request, trying up to RETRIES times. A retryable status waits for the server's Retry-After, if it gives one in - seconds, or SLEEP_TIME. Connection failures are retried after SLEEP_TIME: - for a GET any failure, for a POST only one that happened before the - request was sent. + seconds (up to MAX_RETRY_AFTER), or SLEEP_TIME. Connection failures are + retried after SLEEP_TIME: for a GET any failure, for a POST only one that + happened before the request was sent. Each attempt uses a new connection, + so a POST never goes out on a kept-alive connection the server is closing. """ - for attempt in range(1, RETRIES + 1): - last = attempt == RETRIES - _rewind(kwargs.get("files")) + attempts = max(1, RETRIES) + for attempt in range(1, attempts + 1): + last = attempt == attempts try: - response = self.session.request( + response = requests.request( method, url=f"{self.api_link}/{endpoint}", + headers=self.headers, timeout=timeout, **kwargs, ) @@ -76,12 +83,17 @@ def _request( def _dalton_get(self, endpoint: str) -> requests.Response: return self._request("GET", endpoint, RETRY_CODES_GET, self.timeout) - def _dalton_post(self, endpoint: str, data: dict, files: dict) -> requests.Response: + def _dalton_post(self, endpoint: str, data: dict, files) -> requests.Response: + # read the files once, from where the caller left them, so that every + # attempt sends the same bytes (requests reads them into memory anyway) + pairs = files.items() if hasattr(files, "items") else files + files = [(name, _read_file(name, value)) for name, value in pairs] + connect = self.timeout[0] if isinstance(self.timeout, tuple) else self.timeout return self._request( "POST", endpoint, RETRY_CODES_POST, - (self.timeout, self.submit_timeout), + (connect, self.submit_timeout), data=data, files=files, ) @@ -102,7 +114,8 @@ def submit_job( configuration (str): sensor configuration. Can be read from a yaml file. See "examples/job_submission.py" custom_rules (str): rules to be tested. Can be read from a rules file. See "examples/job_submission.py" files (dict): dictionary that includes tuple with pcap file. Ex. {"coverage-pcap0": ("test_job.pcap", pcap)}. - A list of (name, tuple) pairs also works. Open files are rewound before a retry. + A list of (name, tuple) pairs also works. An open file is read from its current + position, as requests would; pass bytes to submit the same pcap more than once. Returns: str: the job ID, or a comma-separated list of job IDs if Dalton split the pcap. @@ -127,7 +140,7 @@ def submit_job( jids = response.text.strip() if not TEAPOT_JIDS_RE.match(jids): raise DaltonAPIError( - f"Dalton did not accept the job: {_page_text(jids)[:500]}" + f"Dalton did not accept the job: {_error_text(jids)[:500]}" ) return jids @@ -203,31 +216,37 @@ def get_current_sensors(self) -> dict: return response.json() -def _page_text(html: str) -> str: - """The visible text of an HTML page, whitespace collapsed.""" - text = re.sub(r"(?is)<(script|style)\b.*?", " ", html) +def _error_text(page: str) -> str: + """The message on Dalton's error page, or the page's text if it isn't one.""" + match = re.search(r'(?is)
(.*?)
', page) + text = match.group(1) if match else page + text = re.sub(r"(?is)<(script|style)\b.*?", " ", text) text = re.sub(r"<[^>]+>", " ", text) - return " ".join(text.split()) + return " ".join(html.unescape(text).split()) -def _rewind(files) -> None: - """Seek open files back to the start, so a retry sends them whole.""" - if not files: - return - values = files.values() if hasattr(files, "values") else (v for _, v in files) - for value in values: - fileobj = value[1] if isinstance(value, tuple) else value - if hasattr(fileobj, "seek"): - fileobj.seek(0) +def _read_file(name, value): + """A requests files value with any open file replaced by its contents.""" + if isinstance(value, tuple): + if hasattr(value[1], "read"): + return (value[0], value[1].read(), *value[2:]) + return value + if hasattr(value, "read"): + # requests names the upload after the file; keep that name + return (requests.utils.guess_filename(value) or name, value.read()) + return value def _never_sent(exc: Exception) -> bool: - """Whether a request failed while connecting, before the server saw it.""" + """Whether a request failed while connecting, before the server saw it. + + urllib3 raises ProxyError only before the proxy connection is made. + """ reason = getattr(exc.args[0], "reason", None) if exc.args else None - return isinstance(reason, ConnectTimeoutError) + return isinstance(reason, (ConnectTimeoutError, ProxyError)) def _retry_after(response: requests.Response) -> float: """Seconds to wait before retrying: the server's Retry-After, or SLEEP_TIME.""" value = response.headers.get("Retry-After", "").strip() - return int(value) if value.isdigit() else SLEEP_TIME + return min(int(value), MAX_RETRY_AFTER) if value.isdigit() else SLEEP_TIME diff --git a/api/examples/job_submission.py b/api/examples/job_submission.py index f3e5abb..7cf4e5d 100644 --- a/api/examples/job_submission.py +++ b/api/examples/job_submission.py @@ -34,14 +34,15 @@ # since techs and rulesets are a list we use a loop to get the corresponding tech and ruleset # you will probably use this if you want to test with multiple versions of the sensor with open(PCAP, "rb") as pcap: - for tech in dalton_sensor_techs: - # the client rewinds the pcap before each upload, so one open file serves every job - files = {"coverage-pcap0": (os.path.basename(PCAP), pcap)} - jid = dalton_client.submit_job( - sensor_tech=tech, - prod_ruleset=dalton_rulesets[0], - configuration=configuration, - custom_rules=custom_rules, - files=files, - ) - print(f"{tech}: {jid}") + pcap_bytes = pcap.read() +for tech in dalton_sensor_techs: + # bytes can be sent any number of times; an open file would be read to its end by the first job + files = {"coverage-pcap0": (os.path.basename(PCAP), pcap_bytes)} + jid = dalton_client.submit_job( + sensor_tech=tech, + prod_ruleset=dalton_rulesets[0], + configuration=configuration, + custom_rules=custom_rules, + files=files, + ) + print(f"{tech}: {jid}") diff --git a/tests/test_api_client.py b/tests/test_api_client.py index 2b7f5eb..983aa16 100644 --- a/tests/test_api_client.py +++ b/tests/test_api_client.py @@ -34,7 +34,7 @@ def make_response(status=200, text="", headers=None): class FakeRequests: - """Stands in for Session.request: replays responses (or raises) and records calls.""" + """Stands in for requests.request: replays responses (or raises) and records calls.""" def __init__(self, *responses): self.responses = list(responses) @@ -43,7 +43,7 @@ def __init__(self, *responses): def __call__(self, method, url, **kwargs): files = kwargs.get("files") or {} pairs = files.items() if hasattr(files, "items") else files - sent = {name: value[1].read() for name, value in pairs} + sent = {name: value[1] for name, value in pairs} self.calls.append({"method": method, "url": url, **kwargs, "sent": sent}) response = self.responses.pop(0) if isinstance(response, Exception): @@ -62,7 +62,7 @@ def sleeps(monkeypatch): def fake(monkeypatch, sleeps): def install(*responses): fake_requests = FakeRequests(*responses) - monkeypatch.setattr(requests.Session, "request", fake_requests) + monkeypatch.setattr(dalton.requests, "request", fake_requests) return fake_requests return install @@ -110,6 +110,29 @@ def test_submit_timeout_is_configurable(fake): assert requests_.calls[0]["timeout"] == (5, 600) +def test_timeout_tuple(fake): + requests_ = fake(make_response(text="5"), make_response(text=JID)) + client = DaltonAPI("http://dalton.test", timeout=(5, 30)) + client.get_max_pcaps() + submit(client) + assert [call["timeout"] for call in requests_.calls] == [(5, 30), (5, None)] + + +def test_headers_are_sent(fake): + requests_ = fake(make_response(text="5"), make_response(text=JID)) + client = DaltonAPI("http://dalton.test", headers={"X-Api-Key": "k"}) + client.get_max_pcaps() + submit(client) + assert [call["headers"] for call in requests_.calls] == [{"X-Api-Key": "k"}] * 2 + + +def test_no_retries_still_makes_one_attempt(fake, monkeypatch): + monkeypatch.setattr(dalton, "RETRIES", 0) + requests_ = fake(make_response(text="5")) + assert DaltonAPI("http://dalton.test").get_max_pcaps() == 5 + assert len(requests_.calls) == 1 + + def test_submit_split_pcap_job_ids(fake): jids = f"{JID},teapot_fedcba9876543210" fake(make_response(text=jids)) @@ -117,10 +140,16 @@ def test_submit_split_pcap_job_ids(fake): def test_submit_error_page(fake): - page = "

Error: no sensor

" + page = ( + "" + 'Dalton Home 
' + '

Error!

' + '

No sensor & no rules.

' + ) fake(make_response(text=page)) - with pytest.raises(DaltonAPIError, match="Error: no sensor"): + with pytest.raises(DaltonAPIError) as error: submit(DaltonAPI("http://dalton.test")) + assert str(error.value) == "Dalton did not accept the job: No sensor & no rules." def test_get_retries_then_succeeds(fake): @@ -206,14 +235,68 @@ def test_retry_after_is_honoured(fake, sleeps): assert sleeps == [7, dalton.SLEEP_TIME] +def test_retry_after_is_capped(fake, sleeps): + fake(make_response(429, headers={"Retry-After": "86400"}), make_response(text="5")) + DaltonAPI("http://dalton.test").get_max_pcaps() + assert sleeps == [dalton.MAX_RETRY_AFTER] + + +class Pipe: + """A stream that can be read but not seeked, like stdin.""" + + def __init__(self, data): + self.data = data + + def read(self, *args): + data, self.data = self.data, b"" + return data + + +def test_submit_reads_files_from_where_they_are(fake): + positioned = io.BytesIO(b"headerPCAP") + positioned.seek(6) + requests_ = fake(make_response(503), make_response(text=JID)) + files = { + "coverage-pcap0": ("a.pcap", positioned), + "coverage-pcap1": ("b.pcap", Pipe(b"piped")), + } + assert submit(DaltonAPI("http://dalton.test"), files=files) == JID + assert [call["sent"] for call in requests_.calls] == [ + {"coverage-pcap0": b"PCAP", "coverage-pcap1": b"piped"} + ] * 2 + + +def test_submit_bare_file_keeps_its_name(fake, tmp_path): + path = tmp_path / "capture.pcap" + path.write_bytes(b"pcap") + requests_ = fake(make_response(text=JID)) + with open(path, "rb") as pcap: + submit(DaltonAPI("http://dalton.test"), files={"coverage-pcap0": pcap}) + assert requests_.calls[0]["files"] == [ + ("coverage-pcap0", ("capture.pcap", b"pcap")) + ] + + def test_never_sent_with_real_connections(): - # a port nothing listens on refuses the connection - with socket.socket() as probe: - probe.bind(("127.0.0.1", 0)) - closed_port = probe.getsockname()[1] - with pytest.raises(requests.ConnectionError) as refused: - requests.post(f"http://127.0.0.1:{closed_port}/", data={"a": 1}, timeout=5) - assert dalton._never_sent(refused.value) + session = requests.Session() + session.trust_env = False # ignore HTTP_PROXY and friends + # a port that is bound but not listening refuses the connection; keeping it + # bound stops anything else taking it + with socket.socket() as refusing: + refusing.bind(("127.0.0.1", 0)) + refusing_url = "http://127.0.0.1:%d" % refusing.getsockname()[1] + with pytest.raises(requests.ConnectionError) as refused: + session.post(refusing_url + "/", data={"a": 1}, timeout=5) + assert dalton._never_sent(refused.value) + # a proxy that refuses the connection + with pytest.raises(requests.exceptions.ProxyError) as proxy: + session.post( + "http://dalton.test/", + data={"a": 1}, + proxies={"http": refusing_url}, + timeout=5, + ) + assert dalton._never_sent(proxy.value) # a server that reads the request and hangs up without replying server = socket.socket() @@ -229,7 +312,7 @@ def hang_up(): thread.start() port = server.getsockname()[1] with pytest.raises(requests.ConnectionError) as dropped: - requests.post(f"http://127.0.0.1:{port}/", data={"a": 1}, timeout=5) + session.post(f"http://127.0.0.1:{port}/", data={"a": 1}, timeout=5) thread.join() server.close() assert not dalton._never_sent(dropped.value) From 3b080e8f7550ad012d73ebaa60930278928f75b8 Mon Sep 17 00:00:00 2001 From: David Wharton Date: Sat, 14 Nov 2026 15:10:24 +0000 Subject: [PATCH 4/6] API client: fix third-review findings - a POST dropped by a proxy after it was sent is not retried - unresolvable hosts aren't retried; connection failures wait 10s, not 60s - GETs are retried when reading the body fails - ignore non-ASCII digits in Retry-After - files type hint, example docstring, test thread timeouts --- README.rst | 5 +- api/dalton.py | 53 +++++++++++++++----- api/examples/job_submission.py | 4 +- tests/test_api_client.py | 88 ++++++++++++++++++++++++++-------- 4 files changed, 115 insertions(+), 35 deletions(-) diff --git a/README.rst b/README.rst index cc59803..ab68e79 100644 --- a/README.rst +++ b/README.rst @@ -905,8 +905,9 @@ so pass bytes to submit the same pcap more than once. Each request is tried up to three times. A GET is retried after a 429 or 5xx response, a connection error or a timeout. A job submission is retried only when the request can't have -queued the job: a 429 or 503 response, or a failure to connect. Retries wait for the server's -``Retry-After``, if it gives one in seconds (at most 300), or 60 seconds. +queued the job: a 429 or 503 response, or a failure to connect. A retry after a 429 or 5xx waits +for the server's ``Retry-After``, if it gives one in seconds (at most 300), or 60 seconds; a retry +after a connection error waits 10 seconds. A host name that doesn't resolve isn't retried. The client waits up to 60 seconds to connect and for the reply to a GET (``DaltonAPI(url, timeout=...)``). Dalton processes the pcap before it replies to a job submission, so by default diff --git a/api/dalton.py b/api/dalton.py index 5d769c7..9987eca 100644 --- a/api/dalton.py +++ b/api/dalton.py @@ -5,10 +5,14 @@ import time import requests +import urllib3.exceptions from urllib3.exceptions import ConnectTimeoutError, ProxyError RETRIES = 3 SLEEP_TIME = 60 +# wait before retrying a connection failure: long enough for a restart, short +# enough that a wrong port fails quickly +CONNECT_SLEEP_TIME = 10 TIMEOUT = 60 # longest wait a server's Retry-After can ask for MAX_RETRY_AFTER = 300 @@ -52,9 +56,10 @@ def _request( A retryable status waits for the server's Retry-After, if it gives one in seconds (up to MAX_RETRY_AFTER), or SLEEP_TIME. Connection failures are - retried after SLEEP_TIME: for a GET any failure, for a POST only one that - happened before the request was sent. Each attempt uses a new connection, - so a POST never goes out on a kept-alive connection the server is closing. + retried after CONNECT_SLEEP_TIME: for a GET any failure, for a POST only + one that happened before the request was sent. A host name that doesn't + resolve isn't retried. Each attempt uses a new connection, so a POST never + goes out on a kept-alive connection the server is closing. """ attempts = max(1, RETRIES) for attempt in range(1, attempts + 1): @@ -69,10 +74,19 @@ def _request( ) except requests.exceptions.SSLError: raise - except (requests.ConnectionError, requests.Timeout) as exc: - if last or not (method == "GET" or _never_sent(exc)): + except ( + requests.ConnectionError, + requests.Timeout, + requests.exceptions.ChunkedEncodingError, + requests.exceptions.ContentDecodingError, + ) as exc: + if ( + last + or _name_not_resolved(exc) + or not (method == "GET" or _never_sent(exc)) + ): raise - time.sleep(SLEEP_TIME) + time.sleep(CONNECT_SLEEP_TIME) continue if response.status_code in retry_codes and not last: time.sleep(_retry_after(response)) @@ -104,7 +118,7 @@ def submit_job( prod_ruleset: str, configuration: str, custom_rules: str, - files: dict, + files: dict | list, ) -> str: """Submits a job for a specific sensor tech, ruleset, configuration and pcap. @@ -237,16 +251,29 @@ def _read_file(name, value): return value +def _reason(exc: Exception): + """The urllib3 error behind a requests connection error, if there is one.""" + reason = getattr(exc.args[0], "reason", None) if exc.args else None + # a ProxyError can wrap a failure from before or after the request went out + if isinstance(reason, ProxyError): + reason = getattr(reason, "original_error", None) + return reason + + def _never_sent(exc: Exception) -> bool: - """Whether a request failed while connecting, before the server saw it. + """Whether a request failed while connecting, before the server (or proxy) saw it.""" + return isinstance(_reason(exc), ConnectTimeoutError) - urllib3 raises ProxyError only before the proxy connection is made. - """ - reason = getattr(exc.args[0], "reason", None) if exc.args else None - return isinstance(reason, (ConnectTimeoutError, ProxyError)) + +def _name_not_resolved(exc: Exception) -> bool: + """Whether the host name didn't resolve (urllib3 2 only), which a retry won't fix.""" + error = getattr(urllib3.exceptions, "NameResolutionError", None) + return error is not None and isinstance(_reason(exc), error) def _retry_after(response: requests.Response) -> float: """Seconds to wait before retrying: the server's Retry-After, or SLEEP_TIME.""" value = response.headers.get("Retry-After", "").strip() - return min(int(value), MAX_RETRY_AFTER) if value.isdigit() else SLEEP_TIME + if value.isascii() and value.isdigit(): + return min(int(value), MAX_RETRY_AFTER) + return SLEEP_TIME diff --git a/api/examples/job_submission.py b/api/examples/job_submission.py index 7cf4e5d..fe81f42 100644 --- a/api/examples/job_submission.py +++ b/api/examples/job_submission.py @@ -1,4 +1,6 @@ -"""Example on how to submit a job using the Dalton Client API. Mock data is in mocks directory. +"""Example on how to submit a job using the Dalton Client API. + +The engine config and custom rules come from the mocks directory, the pcap from the command line. Run it from the repository root: python -m api.examples.job_submission """ diff --git a/tests/test_api_client.py b/tests/test_api_client.py index 983aa16..8bb95a7 100644 --- a/tests/test_api_client.py +++ b/tests/test_api_client.py @@ -6,6 +6,7 @@ import pytest import requests +import urllib3.exceptions from urllib3.exceptions import MaxRetryError, NewConnectionError, ProtocolError from api import dalton @@ -198,10 +199,41 @@ def test_submit_files_as_a_list(fake): ] * 2 -def test_get_retries_connection_errors(fake): +def test_get_retries_connection_errors(fake, sleeps): requests_ = fake(DROPPED, requests.ReadTimeout(), make_response(text="5")) assert DaltonAPI("http://dalton.test").get_max_pcaps() == 5 assert len(requests_.calls) == 3 + assert sleeps == [dalton.CONNECT_SLEEP_TIME] * 2 + + +def test_get_retries_body_read_errors(fake): + requests_ = fake( + requests.exceptions.ChunkedEncodingError(DROPPED), + requests.exceptions.ContentDecodingError(), + make_response(text="5"), + ) + assert DaltonAPI("http://dalton.test").get_max_pcaps() == 5 + assert len(requests_.calls) == 3 + + +def test_submit_not_retried_when_body_read_fails(fake): + requests_ = fake(requests.exceptions.ChunkedEncodingError(DROPPED)) + with pytest.raises(requests.exceptions.ChunkedEncodingError): + submit(DaltonAPI("http://dalton.test")) + assert len(requests_.calls) == 1 + + +@pytest.mark.skipif( + not hasattr(urllib3.exceptions, "NameResolutionError"), + reason="urllib3 1.x reports DNS failures as NewConnectionError", +) +def test_unresolvable_host_is_not_retried(fake): + error = urllib3.exceptions.NameResolutionError("no-such-host", None, OSError()) + unresolved = requests.ConnectionError(MaxRetryError(None, "/", error)) + requests_ = fake(unresolved) + with pytest.raises(requests.ConnectionError): + DaltonAPI("http://no-such-host").get_max_pcaps() + assert len(requests_.calls) == 1 def test_submit_retried_when_connection_refused(fake): @@ -241,6 +273,12 @@ def test_retry_after_is_capped(fake, sleeps): assert sleeps == [dalton.MAX_RETRY_AFTER] +def test_retry_after_non_ascii_digits(fake, sleeps): + fake(make_response(429, headers={"Retry-After": "\u00b2"}), make_response(text="5")) + DaltonAPI("http://dalton.test").get_max_pcaps() + assert sleeps == [dalton.SLEEP_TIME] + + class Pipe: """A stream that can be read but not seeked, like stdin.""" @@ -298,24 +336,36 @@ def test_never_sent_with_real_connections(): ) assert dalton._never_sent(proxy.value) - # a server that reads the request and hangs up without replying - server = socket.socket() - server.bind(("127.0.0.1", 0)) - server.listen() - - def hang_up(): - conn, _ = server.accept() - conn.recv(65536) - conn.close() - - thread = threading.Thread(target=hang_up, daemon=True) - thread.start() - port = server.getsockname()[1] - with pytest.raises(requests.ConnectionError) as dropped: - session.post(f"http://127.0.0.1:{port}/", data={"a": 1}, timeout=5) - thread.join() - server.close() - assert not dalton._never_sent(dropped.value) + # a server, then a proxy, that reads the request and hangs up without replying + for proxied in (False, True): + with socket.socket() as server: + server.bind(("127.0.0.1", 0)) + server.listen() + server.settimeout(5) + url = "http://127.0.0.1:%d" % server.getsockname()[1] + + def hang_up(): + try: + conn, _ = server.accept() + except OSError: + return + conn.recv(65536) + conn.close() + + thread = threading.Thread(target=hang_up, daemon=True) + thread.start() + with pytest.raises(requests.ConnectionError) as dropped: + if proxied: + session.post( + "http://dalton.test/", + data={"a": 1}, + proxies={"http": url}, + timeout=5, + ) + else: + session.post(url + "/", data={"a": 1}, timeout=5) + thread.join(10) + assert not dalton._never_sent(dropped.value) def test_submit_without_custom_rules(fake): From 8cf6e2b4cad5af58f55ebcb56a191ede7b9971fc Mon Sep 17 00:00:00 2001 From: David Wharton Date: Sat, 14 Nov 2026 15:28:18 +0000 Subject: [PATCH 5/6] API client: fix fourth-review findings - retry a job submission only on 429; a proxy's 503 may follow a queued job - import on Python 3.8/3.9; read files given as lists; accept files=None - one-hour default for the submission reply instead of no limit --- README.rst | 13 +++++++------ api/dalton.py | 25 ++++++++++++++++--------- tests/test_api_client.py | 38 +++++++++++++++++++++++++++++--------- 3 files changed, 52 insertions(+), 24 deletions(-) diff --git a/README.rst b/README.rst index ab68e79..fff86c1 100644 --- a/README.rst +++ b/README.rst @@ -905,14 +905,15 @@ so pass bytes to submit the same pcap more than once. Each request is tried up to three times. A GET is retried after a 429 or 5xx response, a connection error or a timeout. A job submission is retried only when the request can't have -queued the job: a 429 or 503 response, or a failure to connect. A retry after a 429 or 5xx waits -for the server's ``Retry-After``, if it gives one in seconds (at most 300), or 60 seconds; a retry -after a connection error waits 10 seconds. A host name that doesn't resolve isn't retried. +queued the job: a 429 response, or a failure to connect. A retry after a 429 or 5xx waits for the +server's ``Retry-After``, if it gives one in seconds (at most 300), or 60 seconds; a retry after a +connection error waits 10 seconds. With urllib3 2, a host name that doesn't resolve isn't retried. The client waits up to 60 seconds to connect and for the reply to a GET (``DaltonAPI(url, -timeout=...)``). Dalton processes the pcap before it replies to a job submission, so by default -there's no limit on that wait; ``submit_timeout=...`` sets one, but a submission that times out -may still have been queued. ``headers={...}`` adds headers, such as an API key, to every request. +timeout=...)``). Dalton processes the pcap before it replies to a job submission, so the client +waits up to an hour for that reply (``submit_timeout=...``; ``None`` waits forever). A submission +that times out may still have been queued. ``headers={...}`` adds headers, such as an API key, to +every request. Submit Job using the API ------------------------ diff --git a/api/dalton.py b/api/dalton.py index 9987eca..7f7e026 100644 --- a/api/dalton.py +++ b/api/dalton.py @@ -1,5 +1,7 @@ """Dalton API client.""" +from __future__ import annotations + import html import re import time @@ -14,14 +16,17 @@ # enough that a wrong port fails quickly CONNECT_SLEEP_TIME = 10 TIMEOUT = 60 +# Dalton processes the pcap before it answers a job submission; this only +# stops the client waiting forever +SUBMIT_TIMEOUT = 3600 # longest wait a server's Retry-After can ask for MAX_RETRY_AFTER = 300 # GETs are retried on any of these. A job submission (POST) is retried only on -# the codes that mean the request wasn't processed, so a retry can't create a -# second job. +# a 429, which means the request wasn't processed, so a retry can't create a +# second job (a proxy's 503 can come after Dalton queued it). RETRY_CODES_GET = (429, 500, 502, 503, 504) -RETRY_CODES_POST = (429, 503) +RETRY_CODES_POST = (429,) # what a teapot job submission returns: a job ID, or several, comma-separated, # when the pcap was split @@ -33,15 +38,17 @@ class DaltonAPIError(Exception): class DaltonAPI: - def __init__(self, api_link, timeout=TIMEOUT, submit_timeout=None, headers=None): + def __init__( + self, api_link, timeout=TIMEOUT, submit_timeout=SUBMIT_TIMEOUT, headers=None + ): """ Args: api_link (str): Dalton base URL. Ex. http://localhost timeout (float): seconds to wait for a connection, and for a reply to a GET. A (connect, read) tuple sets the two separately. submit_timeout (float): seconds to wait for the reply to a job submission. - Dalton processes the pcap before it replies, so the default is no limit; - a submission that times out may still have been queued. + Dalton processes the pcap before it replies, so the default is an hour; + None waits forever. A submission that times out may still have been queued. headers (dict): extra headers sent with every request, e.g. an API key """ self.api_link = api_link @@ -100,7 +107,7 @@ def _dalton_get(self, endpoint: str) -> requests.Response: def _dalton_post(self, endpoint: str, data: dict, files) -> requests.Response: # read the files once, from where the caller left them, so that every # attempt sends the same bytes (requests reads them into memory anyway) - pairs = files.items() if hasattr(files, "items") else files + pairs = files.items() if hasattr(files, "items") else (files or []) files = [(name, _read_file(name, value)) for name, value in pairs] connect = self.timeout[0] if isinstance(self.timeout, tuple) else self.timeout return self._request( @@ -241,8 +248,8 @@ def _error_text(page: str) -> str: def _read_file(name, value): """A requests files value with any open file replaced by its contents.""" - if isinstance(value, tuple): - if hasattr(value[1], "read"): + if isinstance(value, (tuple, list)): + if len(value) > 1 and hasattr(value[1], "read"): return (value[0], value[1].read(), *value[2:]) return value if hasattr(value, "read"): diff --git a/tests/test_api_client.py b/tests/test_api_client.py index 8bb95a7..30e1586 100644 --- a/tests/test_api_client.py +++ b/tests/test_api_client.py @@ -101,8 +101,8 @@ def test_submit_returns_job_id_and_posts_once(fake): assert data["teapotJob"] == 1 assert data["optionCustomRuleset"] == "custom" assert data["custom_ruleset"].startswith("alert ") - # Dalton processes the pcap before replying: no read timeout by default - assert requests_.calls[0]["timeout"] == (dalton.TIMEOUT, None) + # Dalton processes the pcap before replying: a long read timeout by default + assert requests_.calls[0]["timeout"] == (dalton.TIMEOUT, dalton.SUBMIT_TIMEOUT) def test_submit_timeout_is_configurable(fake): @@ -116,7 +116,10 @@ def test_timeout_tuple(fake): client = DaltonAPI("http://dalton.test", timeout=(5, 30)) client.get_max_pcaps() submit(client) - assert [call["timeout"] for call in requests_.calls] == [(5, 30), (5, None)] + assert [call["timeout"] for call in requests_.calls] == [ + (5, 30), + (5, dalton.SUBMIT_TIMEOUT), + ] def test_headers_are_sent(fake): @@ -174,16 +177,18 @@ def test_not_found_is_not_retried(fake): def test_submit_retry_resends_the_whole_pcap(fake): - requests_ = fake(make_response(503), make_response(text=JID)) + requests_ = fake(make_response(429), make_response(text=JID)) assert submit(DaltonAPI("http://dalton.test"), pcap=b"0123456789") == JID assert [call["sent"]["coverage-pcap0"] for call in requests_.calls] == [ b"0123456789" ] * 2 -def test_submit_not_retried_when_it_may_have_run(fake): - # a 500 may come after the job was queued; retrying could submit it twice - requests_ = fake(make_response(500)) +@pytest.mark.parametrize("status", [500, 502, 503, 504]) +def test_submit_not_retried_when_it_may_have_run(fake, status): + # a 5xx (even a proxy's 503) may come after the job was queued; retrying + # could submit it twice + requests_ = fake(make_response(status)) with pytest.raises(requests.HTTPError): submit(DaltonAPI("http://dalton.test")) assert len(requests_.calls) == 1 @@ -191,7 +196,7 @@ def test_submit_not_retried_when_it_may_have_run(fake): def test_submit_files_as_a_list(fake): pcap = io.BytesIO(b"0123456789") - requests_ = fake(make_response(503), make_response(text=JID)) + requests_ = fake(make_response(429), make_response(text=JID)) files = [("coverage-pcap0", ("test.pcap", pcap))] assert submit(DaltonAPI("http://dalton.test"), files=files) == JID assert [call["sent"]["coverage-pcap0"] for call in requests_.calls] == [ @@ -293,7 +298,7 @@ def read(self, *args): def test_submit_reads_files_from_where_they_are(fake): positioned = io.BytesIO(b"headerPCAP") positioned.seek(6) - requests_ = fake(make_response(503), make_response(text=JID)) + requests_ = fake(make_response(429), make_response(text=JID)) files = { "coverage-pcap0": ("a.pcap", positioned), "coverage-pcap1": ("b.pcap", Pipe(b"piped")), @@ -378,3 +383,18 @@ def test_submit_without_custom_rules(fake): files={"coverage-pcap0": ("test.pcap", io.BytesIO(b"x"))}, ) assert "optionCustomRuleset" not in requests_.calls[0]["data"] + + +def test_submit_file_in_a_list_value_is_read_once(fake): + requests_ = fake(make_response(429), make_response(text=JID)) + files = {"coverage-pcap0": ["a.pcap", io.BytesIO(b"pcap")]} + assert submit(DaltonAPI("http://dalton.test"), files=files) == JID + assert [call["sent"]["coverage-pcap0"] for call in requests_.calls] == [b"pcap"] * 2 + + +def test_submit_without_files(fake): + page = '

You must specify a PCAP file.

' + requests_ = fake(make_response(text=page)) + with pytest.raises(DaltonAPIError, match="You must specify a PCAP file"): + DaltonAPI("http://dalton.test").submit_job("s", "r", "c", "", None) + assert requests_.calls[0]["files"] == [] From d0018d0a55efd4b9faffaafc514de7e857a04f23 Mon Sep 17 00:00:00 2001 From: David Wharton Date: Sat, 14 Nov 2026 15:32:36 +0000 Subject: [PATCH 6/6] API client: submit_job docstring says files can be a list --- api/dalton.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/api/dalton.py b/api/dalton.py index 7f7e026..e50ca34 100644 --- a/api/dalton.py +++ b/api/dalton.py @@ -134,8 +134,8 @@ def submit_job( prod_ruleset (str): path to prod ruleset. Ex. /opt/dalton/rulesets/suricata/ET-20230301-all-suricata.rules configuration (str): sensor configuration. Can be read from a yaml file. See "examples/job_submission.py" custom_rules (str): rules to be tested. Can be read from a rules file. See "examples/job_submission.py" - files (dict): dictionary that includes tuple with pcap file. Ex. {"coverage-pcap0": ("test_job.pcap", pcap)}. - A list of (name, tuple) pairs also works. An open file is read from its current + files (dict | list): the pcap files, as a dict or a list of (name, value) pairs. + Ex. {"coverage-pcap0": ("test_job.pcap", pcap)}. An open file is read from its current position, as requests would; pass bytes to submit the same pcap more than once. Returns: