Ignore specific url parameters when following location headers (#3856)

Co-authored-by: Jan Klopper <[email protected]>
Co-authored-by: ammar92 <[email protected]>

boefjes/boefjes/plugins/kat_answer_parser/normalize.py

@@ -10,4 +10,4 @@ def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
 
     bit_id = data["schema"].removeprefix("/bit/")
 
-    yield Config(ooi=input_ooi["primary_key"], bit_id=bit_id, config=data["answer"])
+    yield Config(ooi=data["answer_ooi"], bit_id=bit_id, config=data["answer"])

boefjes/tests/plugins/test_answer_parser.py

@@ -13,10 +13,10 @@ def test_config_yielded(normalizer_runner):
         normalizer_runner.run(meta, bytes(raw, "UTF-8"))
 
     with pytest.raises(ValidationError):
-        raw = '{"schema": "/bit/port-classification-ip", "answer": [{"key": "test"}]}'
+        raw = '{"schema": "/bit/port-classification-ip", "answer": [{"key": "test"}], "answer_ooi": "Network|internet"}'
         normalizer_runner.run(meta, bytes(raw, "UTF-8"))
 
-    raw = '{"schema": "/bit/port-classification-ip", "answer": {"key": "test"}}'
+    raw = '{"schema": "/bit/port-classification-ip", "answer": {"key": "test"}, "answer_ooi": "Network|internet"}'
     output = normalizer_runner.run(meta, bytes(raw, "UTF-8"))
 
     assert len(output.observations) == 1

octopoes/bits/ask_url_params_to_ignore/ask_url_params_to_ignore.py

@@ -0,0 +1,17 @@
+import json
+from collections.abc import Iterator
+from pathlib import Path
+from typing import Any
+
+from octopoes.models import OOI
+from octopoes.models.ooi.network import Network
+from octopoes.models.ooi.question import Question
+
+
+def run(input_ooi: Network, additional_oois: list, config: dict[str, Any]) -> Iterator[OOI]:
+    network = input_ooi
+
+    with (Path(__file__).parent / "question_schema.json").open() as f:
+        schema = json.load(f)
+
+    yield Question(ooi=network.reference, schema_id=schema["$id"], json_schema=json.dumps(schema))

octopoes/bits/ask_url_params_to_ignore/bit.py

@@ -0,0 +1,10 @@
+from bits.definitions import BitDefinition
+from octopoes.models.ooi.network import Network
+
+BIT = BitDefinition(
+    id="ask_url_params_to_ignore",
+    consumes=Network,
+    parameters=[],
+    min_scan_level=0,
+    module="bits.ask_url_params_to_ignore.ask_url_params_to_ignore",
+)

octopoes/bits/ask_url_params_to_ignore/question_schema.json

@@ -0,0 +1,17 @@
+{
+  "$schema": "https://json-schema.org/draft/2019-09/schema",
+  "$id": "/bit/oois-in-headers",
+  "type": "object",
+  "default": {},
+  "required": [
+    "ignored_url_parameters"
+  ],
+  "properties": {
+    "ignored_url_parameters": {
+      "description": "Comma separated list of url parameters that are ignored when following location headers.",
+      "type": "string",
+      "pattern": "^(\\s*(,*)[^,]+,?\\s*)*$",
+      "default": "session_id, phpsessid, jsessionid, cifd, cftoken, asp.net_sessionid"
+    }
+  }
+}

octopoes/bits/oois_in_headers/bit.py

@@ -2,5 +2,9 @@
 from octopoes.models.types import HTTPHeader
 
 BIT = BitDefinition(
-    id="oois-in-headers", consumes=HTTPHeader, parameters=[], module="bits.oois_in_headers.oois_in_headers"
+    id="oois-in-headers",
+    consumes=HTTPHeader,
+    parameters=[],
+    module="bits.oois_in_headers.oois_in_headers",
+    config_ooi_relation_path="HTTPHeader.resource.website.hostname.network",
 )

octopoes/bits/oois_in_headers/oois_in_headers.py

@@ -1,7 +1,7 @@
 import re
 from collections.abc import Iterator
 from typing import Any
-from urllib.parse import urljoin, urlparse
+from urllib.parse import parse_qs, urlencode, urljoin, urlparse, urlunparse
 
 from pydantic import ValidationError
 
@@ -16,12 +16,33 @@ def is_url(input_str):
     return bool(result.scheme)
 
 
+def get_ignored_url_params(config: dict, config_key: str, default: list) -> list[str]:
+    ignored_url_params = config.get(config_key)
+    if ignored_url_params is None:
+        return default
+    return [param.strip() for param in ignored_url_params.split(",")] if ignored_url_params else []
+
+
+def remove_ignored_params(url: str, ignored_params: list[str]) -> str:
+    parsed_url = urlparse(url)
+    query_params = parse_qs(parsed_url.query)
+    if not query_params:
+        return url
+    filtered_params = {k: v for k, v in query_params.items() if k.lower() not in ignored_params}
+    new_query = urlencode(filtered_params, doseq=True)
+    new_url = urlunparse(
+        (parsed_url.scheme, parsed_url.netloc, parsed_url.path, parsed_url.params, new_query, parsed_url.fragment)
+    )
+    return new_url
+
+
 def run(input_ooi: HTTPHeader, additional_oois: list, config: dict[str, Any]) -> Iterator[OOI]:
     network = Network(name="internet")
 
     if input_ooi.key.lower() == "location":
+        ignored_url_params = get_ignored_url_params(config, "ignored_url_parameters", [])
         if is_url(input_ooi.value):
-            u = URL(raw=input_ooi.value, network=network.reference)
+            u = URL(raw=remove_ignored_params(input_ooi.value, ignored_url_params), network=network.reference)
         else:
             # url is not a url but a relative path
             http_url = input_ooi.reference.tokenized.resource.web_url

octopoes/tests/integration/test_ooi_deletion.py

@@ -131,9 +131,9 @@ def test_events_created_in_worker_during_handling(
         xtdb_octopoes_service.process_event(event)
     xtdb_octopoes_service.commit()
 
-    assert len(event_manager.queue) == 7  # Handling OOI delete event triggers Origin delete event
+    assert len(event_manager.queue) == 8  # Handling OOI delete event triggers Origin delete event
 
-    event = event_manager.queue[6]  # OOID]elete event
+    event = event_manager.queue[7]  # OOIDelete event
 
     assert isinstance(event, OriginDBEvent)
     assert event.operation_type.value == "delete"
@@ -229,7 +229,7 @@ def test_deletion_events_after_nxdomain(
     event_manager.complete_process_events(xtdb_octopoes_service)
 
     assert len(list(filter(lambda x: x.operation_type.value == "delete", event_manager.queue))) == 0
-    assert xtdb_octopoes_service.ooi_repository.list_oois({OOI}, valid_time).count == 7
+    assert xtdb_octopoes_service.ooi_repository.list_oois({OOI}, valid_time).count == 8
 
     nxd = NXDOMAIN(hostname=hostname.reference)
     xtdb_octopoes_service.ooi_repository.save(nxd, valid_time)
@@ -250,7 +250,7 @@ def test_deletion_events_after_nxdomain(
     event_manager.complete_process_events(xtdb_octopoes_service)
 
     assert len(list(filter(lambda x: x.operation_type.value == "delete", event_manager.queue))) >= 3
-    assert xtdb_octopoes_service.ooi_repository.list_oois({OOI}, valid_time).count == 5
+    assert xtdb_octopoes_service.ooi_repository.list_oois({OOI}, valid_time).count == 6
 
 
 @pytest.mark.xfail(reason="Wappalyzer works on wrong input objects (to be addressed)")

rocky/rocky/views/ooi_detail.py

@@ -51,7 +51,10 @@ def answer_ooi_questions(self) -> None:
                 messages.error(self.request, error.message)
             return
 
-        self.bytes_client.upload_raw(schema_answer.encode(), {"answer", f"{self.ooi.schema_id}"}, self.ooi.ooi)
+        raw = json.dumps(
+            {"schema": self.ooi.schema_id, "answer": parsed_schema_answer, "answer_ooi": self.ooi.ooi}
+        ).encode()
+        self.bytes_client.upload_raw(raw, {"answer"}, self.ooi.primary_key)
         messages.success(self.request, _("Question has been answered."))
 
     def start_boefje_scan(self) -> None: