diff --git a/documentation/export_mod/owasp_crs_export.md b/documentation/export_mod/owasp_crs_export.md new file mode 100644 index 000000000..c3a36a4ac --- /dev/null +++ b/documentation/export_mod/owasp_crs_export.md @@ -0,0 +1,136 @@ +# OWASP CRS export + +`owasp_crs_export` converts IDS-marked MISP event and object attributes into a +ModSecurity `.conf` fragment to include alongside OWASP CRS. It generates custom +phase 1 rules that log and deny matching requests with HTTP 403. These rules act +directly, independently of CRS anomaly scoring and thresholds. They do not modify +CRS, load configuration, resolve DNS or contact an external service. + +## Supported attributes + +| MISP type | ModSecurity match | +| --- | --- | +| `ip-src` | `REMOTE_ADDR` with `@ipMatch`: client IPv4/IPv6 address or CIDR network | +| `ip-src\|port` | Client address/network **and** `REMOTE_PORT`, joined in a rule chain | +| `domain` | `REQUEST_HEADERS:Host`: domain and its subdomains, optionally followed by a port | +| `hostname` | `REQUEST_HEADERS:Host`: exact hostname, optionally followed by a port | +| `hostname\|port` | Exact hostname and **explicit** port in the Host header | +| `url` | Exact HTTP(S) authority **and** raw request path/query, joined in a rule chain | +| `uri` | Exact origin-form `REQUEST_URI_RAW` (path and optional query) on any host | +| `user-agent` | Exact, case-sensitive `REQUEST_HEADERS:User-Agent` | + +Only attributes with `to_ids=true`, `1`, `"true"` or `"1"` are exported. Boolean +strings are case-insensitive. Missing/false flags are excluded, and there is no +override to include non-IDS attributes. The module never changes input flags. +Deleted events, objects and attributes are excluded. Each object attribute is +processed independently; separate attributes are not combined into an inferred +connection. Composite values must validate completely before a rule is emitted. + +`REMOTE_ADDR` and `REMOTE_PORT` describe the client connection visible to the +ModSecurity connector. Behind another proxy, these may identify that proxy. +The exporter does not trust `X-Forwarded-For` or configure real-client-IP handling. +For source port composites, the port is the client's source port, not the +server's listening port. + +Host rules match the requested HTTP authority; they do not inspect DNS answers +or resolve a destination address. `ip-dst`, `ip-dst|port` and `domain|ip` are +unsupported: `SERVER_ADDR` is the local server address and cannot reliably +represent a forward proxy's upstream destination. Standalone ports, hashes, +email addresses and other types are also skipped. + +## Matching and validation + +Addresses and networks are normalized. DNS names must be qualified and are +normalized to lowercase ASCII IDNA; Host matching ignores case and allows a +trailing DNS dot. Domain matches have label boundaries, so `example.org` does not +match `evilexample.org` or `example.org.other.test`. Ports range from 1 to 65535. +`hostname|port` requires an explicit `:port` in Host; it does not infer omitted +default ports from the proxy's local listening socket. + +URLs must be absolute HTTP(S) without credentials or scoped IPv6 addresses. +Default ports may be absent from Host; non-default ports must be explicit. Empty +paths become `/`, and fragments are removed because clients do not send them. +Paths and queries retain their case, encoding and query order. Empty query +delimiters are preserved. Both origin-form (`/path?query`) and absolute-form +(`https://example.org/path?query`) request targets are accepted. Absolute-form +matches retain the indicator's scheme. Origin-form requests carry no scheme, so +these URL rules cannot distinguish HTTP from HTTPS for that form. They match +the same authority/path/query on either transport. They do not read forwarded +scheme headers. For IP-literal URLs, the Host text uses the normalized IP +representation (bracketed for IPv6); alternative IPv6 spellings are not expanded. + +Exact raw matching does not decode percent escapes, normalize paths or fold +query order. Equivalent application URLs may require separate indicators. +`uri` accepts an ASCII origin-form target beginning with `/`, without a fragment. +URL paths/queries and URIs must use valid ASCII URI characters and percent +escapes; non-ASCII text must be percent-encoded. User agents may contain printable +ASCII, including spaces. Control characters and malformed values are rejected. +Indicator literals are escaped for both ModSecurity's tokenizer and PCRE, +including quotes, regex metacharacters and `%{...}` macro syntax. Each rule uses +`t:none` to reset inherited transformations. + +HTTPS CONNECT tunnels hide request headers and paths without separately +configured TLS inspection. URL rules are not expanded into host-wide blocks. +The exporter does not configure TLS inspection, a proxy or CRS itself. + +## Module API + +Send a request to the misp-modules `/query` endpoint: + +```json +{ + "module": "owasp_crs_export", + "config": {"rule_id_start": "10000"}, + "data": [ + { + "Event": { + "Attribute": [ + {"type": "ip-src", "value": "192.0.2.0/24", "to_ids": true}, + {"type": "url", "value": "https://example.org/payload", "to_ids": true}, + {"type": "domain", "value": "context.example", "to_ids": false} + ] + } + } + ] +} +``` + +`data` accepts a single event or a list, bare events, standard `Event` wrappers, +and legacy exports with `Attribute`/`Object` beside `Event`. Object attributes +are included. The module is discovered automatically and needs no credentials +or additional Python dependencies. + +The response's `data` is a base64-encoded ASCII configuration file. +`summary.count` counts unique rules (a complete chain counts as one), and +`summary.skipped` counts non-IDS, deleted, unsupported and malformed attributes. +Whole deleted events/objects are ignored before counting their attributes. +Warnings explain skipped values and integration limitations. Invalid request +structure, invalid flags or invalid configuration returns an `error` without a +file. Empty events produce comments only. + +Rules are deduplicated and sorted before assigning consecutive IDs. +`rule_id_start` defaults to `10000` and accepts an integer or decimal string from +`10000` to `99999`. IDs stay within this local range, outside CRS's `900000-999999` +range; exports that would exceed `99999` fail without a file. A chain uses one +ID. Reserve the entire used range across all loaded custom rules. Identical +inputs produce identical output, but adding/removing an indicator can shift IDs. +Replace the previous fragment rather than including multiple generations. Use +non-overlapping ranges for separately loaded exports. + +## Integration + +Decode the response to `/etc/modsecurity/misp-ioc.conf`. Include it after your +ModSecurity engine configuration and `crs-setup.conf`, before the CRS rule files +and before any local rule that disables the engine or skips request inspection: + +```apache +Include /etc/modsecurity/crs/crs-setup.conf +Include /etc/modsecurity/misp-ioc.conf +Include /etc/modsecurity/crs/rules/*.conf +``` + +Adapt paths and include syntax to your connector. The generated fragment does +not set `SecRuleEngine`: `On` is required to block, `DetectionOnly` evaluates +and logs without blocking, and `Off` disables inspection. Existing exceptions, +rule removals and engine controls still apply. Validate the complete deployed +configuration before reloading it, for example with `apachectl -t` or `nginx -t`. diff --git a/misp_modules/modules/export_mod/owasp_crs_export.py b/misp_modules/modules/export_mod/owasp_crs_export.py new file mode 100644 index 000000000..7ea1e2b5b --- /dev/null +++ b/misp_modules/modules/export_mod/owasp_crs_export.py @@ -0,0 +1,271 @@ +"""Export IDS-marked HTTP indicators as custom ModSecurity rules for OWASP CRS.""" + +import base64 +import ipaddress +import json +import re +from collections import Counter +from urllib.parse import urlsplit + +moduleinfo = { + "version": "1.0", + "author": "MISP Project", + "name": "OWASP CRS Export", + "description": "Export IDS-marked proxy and HTTP indicators as custom ModSecurity rules alongside OWASP CRS.", + "module-type": ["export"], + "logo": "", + "requirements": [], + "features": ( + "Exports validated source addresses/networks, source address/port pairs, domains, hostnames, hostname/port" + " pairs, HTTP(S) URLs, request URIs and user agents. Only to_ids attributes are included; deleted data is" + " excluded. Produces deterministic, deduplicated phase 1 deny rules with configurable local rule IDs." + " No network access or additional dependency is required. See the [usage" + " guide](https://github.com/MISP/misp-modules/blob/main/documentation/export_mod/owasp_crs_export.md)." + ), + "references": [ + "https://coreruleset.org/", + "https://github.com/owasp-modsecurity/ModSecurity/wiki/Reference-Manual-(v2.x)", + ], + "input": "MISP events and object attributes in standard, bare or legacy export format.", + "output": "Base64-encoded ModSecurity .conf fragment; skipped counts and integration warnings in the response.", +} +moduleconfig = ["rule_id_start"] +SUPPORTED_TYPES = {"ip-src", "ip-src|port", "domain", "hostname", "hostname|port", "url", "uri", "user-agent"} + + +def _boolean(value, name): + if value is None or value == "": + return False + if isinstance(value, (bool, int)) and value in (0, 1): + return bool(value) + if isinstance(value, str) and value.lower() in ("true", "false", "1", "0"): + return value.lower() in ("true", "1") + raise ValueError(f"{name} must be true, false, 1 or 0.") + + +def _records(container, key): + records = container.get(key, []) + if not isinstance(records, list) or any(not isinstance(record, dict) for record in records): + raise ValueError(f"{key} must be a list of objects.") + return records + + +def _attributes(data): + if isinstance(data, dict): + data = [data] + if not isinstance(data, list): + raise ValueError("data must be an event object or a list of events.") + for envelope in data: + if not isinstance(envelope, dict): + raise ValueError("Each event must be an object.") + event = envelope.get("Event", envelope) + if not isinstance(event, dict): + raise ValueError("Event must be an object.") + if _boolean(event.get("deleted"), "deleted"): + continue + collections = dict(event, **envelope) + yield from _records(collections, "Attribute") + for obj in _records(collections, "Object"): + if not _boolean(obj.get("deleted"), "deleted"): + yield from _records(obj, "Attribute") + + +def _rule_id_start(config): + if config is None: + config = {} + if not isinstance(config, dict): + raise ValueError("config must be an object.") + value = config.get("rule_id_start", 10000) + if value is None or value == "": + value = 10000 + if isinstance(value, bool) or not isinstance(value, (str, int)) or not re.fullmatch(r"[0-9]+", str(value)): + raise ValueError("rule_id_start must be an integer from 10000 to 99999.") + number = int(value) + if not 10000 <= number <= 99999: + raise ValueError("rule_id_start must be an integer from 10000 to 99999.") + return number + + +def _hostname(value): + name = value.removesuffix(".").encode("idna").decode("ascii").lower() + labels = name.split(".") + if ( + len(name) > 253 + or len(labels) < 2 + or not re.search(r"[a-z]", labels[-1]) + or any(not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", label) for label in labels) + ): + raise ValueError("Invalid qualified DNS name.") + return name + + +def _port(value): + if not re.fullmatch(r"[0-9]{1,5}", value) or not 1 <= int(value) <= 65535: + raise ValueError("Port must be an integer from 1 to 65535.") + return str(int(value)) + + +def _literal(value): + # PCRE hex escapes protect both the configuration tokenizer and regex parser. + # In particular, no quotes, action delimiters or %{...} macros reach SecRule. + return "".join( + char if char.isascii() and (char.isalnum() or char in "_/-:") else f"\\x{ord(char):02x}" for char in value + ) + + +def _host_pattern(host, subdomains=False, port=None, optional_port=False): + prefix = r"(?:[a-z0-9-]+\.)*" if subdomains else "" + if port is None: + suffix = r"(?::[0-9]{1,5})?" + else: + suffix = rf"(?::{port})?" if optional_port else f":{port}" + return rf"(?i:{prefix}{_literal(host)}\.?{suffix})" + + +def _rx(target, pattern): + return target, rf"@rx \A{pattern}\z" + + +def _uri(value): + if not re.fullmatch(r"/[A-Za-z0-9._~:/?\[\]@!$&'()*+,;=%-]*", value) or re.search(r"%(?![0-9a-fA-F]{2})", value): + raise ValueError("URI must be an ASCII origin-form request target with valid percent escapes and no fragment.") + return value + + +def _url(value): + parsed = urlsplit(value) + if parsed.scheme not in ("http", "https") or not parsed.hostname or parsed.username is not None: + raise ValueError("URL must be absolute HTTP(S) without credentials.") + host = parsed.hostname + if "%" in host or parsed.netloc.endswith(":"): + raise ValueError("Scoped addresses and empty ports are unsupported.") + try: + address = ipaddress.ip_address(host) + except ValueError: + if parsed.netloc.startswith("["): + raise ValueError("Only IPv6 addresses may use bracketed URL hosts.") + host = _hostname(host) + else: + host = f"[{address}]" if address.version == 6 else str(address) + port = _port(str(parsed.port)) if parsed.port is not None else ("80" if parsed.scheme == "http" else "443") + authority = _host_pattern(host, port=port, optional_port=port == ("80" if parsed.scheme == "http" else "443")) + path = _uri((parsed.path or "/") + ("?" + parsed.query if parsed.query or "?" in value.split("#", 1)[0] else "")) + # Raw origin-form and absolute-form targets retain path/query case and encoding. + target = rf"(?:(?i:{parsed.scheme})://{authority})?{_literal(path)}" + return (_rx("REQUEST_HEADERS:Host", authority), _rx("REQUEST_URI_RAW", target)) + + +def _matches(kind, value): + if not isinstance(value, str) or not value or any(ord(char) < 32 or ord(char) == 127 for char in value): + raise ValueError("Indicator must be a non-empty string without control characters.") + if kind == "user-agent": + if not value.isascii(): + raise ValueError("User agent must be ASCII.") + return (_rx("REQUEST_HEADERS:User-Agent", _literal(value)),) + if any(char.isspace() for char in value): + raise ValueError("Indicator must not contain whitespace.") + if kind == "url": + return _url(value) + if kind == "uri": + return (_rx("REQUEST_URI_RAW", _literal(_uri(value))),) + port = None + if kind.endswith("|port"): + value, separator, port = value.partition("|") + if not separator: + raise ValueError("Missing host/port separator.") + port = _port(port) + if kind in ("domain", "hostname", "hostname|port"): + return (_rx("REQUEST_HEADERS:Host", _host_pattern(_hostname(value), kind == "domain", port)),) + if "%" in value: + raise ValueError("Scoped IPv6 is unsupported.") + address = ipaddress.ip_network(value, strict=False) if "/" in value else ipaddress.ip_address(value) + match = (("REMOTE_ADDR", f"@ipMatch {address}"),) + return match + (("REMOTE_PORT", f"@eq {port}"),) if port else match + + +def _render(matches, start): + lines = [ + "# MISP OWASP CRS export: IDS-marked HTTP indicators", + "# Custom ModSecurity rules; reserve the generated IDs and replace this fragment on each export.", + "# Requires SecRuleEngine On to block; DetectionOnly logs without blocking.", + ] + if start + len(matches) - 1 > 99999: + raise ValueError("Export exceeds the available local rule ID range (10000-99999).") + for rule_id, conditions in enumerate(sorted(matches), start): + actions = ( + f"id:{rule_id},phase:1,deny,status:403,log,t:none,msg:'MISP IDS" + " indicator',tag:'misp-ioc',severity:'CRITICAL'" + ) + for index, (target, operator) in enumerate(conditions): + action = actions if index == 0 else "t:none" + if index < len(conditions) - 1: + action += ",chain" + lines.append(f'SecRule {target} "{operator}" "{action}"') + return ("\n".join(lines) + "\n").encode("ascii") + + +def handler(q=False): + if not q: + return False + try: + request = json.loads(q) + if not isinstance(request, dict): + raise ValueError("Request must be an object.") + start = _rule_id_start(request.get("config")) + matches, skipped = set(), Counter() + for attribute in _attributes(request.get("data")): + if _boolean(attribute.get("deleted"), "deleted"): + skipped["deleted"] += 1 + continue + if not _boolean(attribute.get("to_ids"), "to_ids"): + skipped["non-IDS"] += 1 + continue + kind = attribute.get("type") + if not isinstance(kind, str) or kind not in SUPPORTED_TYPES: + skipped["unsupported"] += 1 + continue + try: + matches.add(_matches(kind, attribute.get("value"))) + except (ValueError, UnicodeError): + skipped["malformed"] += 1 + output = _render(matches, start) + except (ValueError, TypeError) as exc: + return {"error": f"OWASP CRS export: {exc}"} + return { + "response": [], + "data": base64.b64encode(output).decode("ascii"), + "summary": {"count": len(matches), "skipped": dict(skipped)}, + "warnings": [ + *(f"Skipped {count} {reason} attribute(s)." for reason, count in sorted(skipped.items())), + ( + "Reserve generated rule IDs across all loaded files. Rules directly deny independently of CRS anomaly" + " scores." + ), + "REMOTE_ADDR/REMOTE_PORT describe the client visible to ModSecurity; forwarded headers are not trusted.", + "Host matches inspect the requested authority, not an upstream DNS lookup or resolved destination IP.", + ( + "URL rules require visible HTTP headers and request targets; HTTPS CONNECT tunnels hide paths without" + " TLS inspection. Origin-form URL matches cannot distinguish HTTP from HTTPS; explicit hostname ports" + " must appear in Host." + ), + ], + } + + +def introspection(): + return { + "input": sorted(SUPPORTED_TYPES), + "inputSource": ["event", "event-collection"], + "outputFileExtension": "conf", + "responseType": "text/plain", + "userConfig": { + "rule_id_start": { + "type": "String", + "message": "First local ModSecurity rule ID (10000-99999; default 10000)", + } + }, + } + + +def version(): + return dict(moduleinfo, config=moduleconfig) diff --git a/tests/test_owasp_crs_export.py b/tests/test_owasp_crs_export.py new file mode 100644 index 000000000..dbb8cfebe --- /dev/null +++ b/tests/test_owasp_crs_export.py @@ -0,0 +1,340 @@ +import base64 +import copy +import ctypes +import ctypes.util +import json +import re + +import pytest +from tornado.testing import AsyncHTTPTestCase +from tornado.web import Application + +from misp_modules import __main__ as service +from misp_modules.modules.export_mod import owasp_crs_export as export + + +def attribute(kind="ip-src", value="192.0.2.1", **metadata): + return {"type": kind, "value": value, "to_ids": True, **metadata} + + +def run(attributes=None, *, data=None, config=None): + result = export.handler( + json.dumps({"data": {"Attribute": attributes or []} if data is None else data, "config": config}) + ) + assert "error" not in result, result + return base64.b64decode(result["data"], validate=True).decode("ascii"), result + + +@pytest.mark.parametrize("flag", [True, 1, "true", "TRUE", "1"]) +def test_only_explicit_ids_flags_are_exported(flag): + _, result = run([attribute(to_ids=flag)]) + assert result["summary"] == {"count": 1, "skipped": {}} + + +@pytest.mark.parametrize("flag", [False, 0, "false", "FALSE", "0", None, ""]) +def test_false_ids_flags_are_excluded(flag): + text, result = run([attribute(to_ids=flag)]) + assert "SecRule " not in text + assert result["summary"] == {"count": 0, "skipped": {"non-IDS": 1}} + + +def test_missing_ids_deleted_events_objects_and_attributes(): + missing = attribute() + del missing["to_ids"] + data = [ + {"Event": {"Attribute": [missing, attribute(deleted="true")]}}, + {"Event": {"deleted": True, "Attribute": [attribute()]}}, + {"Object": [{"deleted": "1", "Attribute": [attribute()]}]}, + {"Object": [{"Attribute": [attribute(to_ids="0")]}]}, + ] + text, result = run(data=data) + assert "SecRule " not in text + assert result["summary"] == {"count": 0, "skipped": {"non-IDS": 2, "deleted": 1}} + + +def test_event_shapes_deduplication_order_and_no_mutation(): + data = [ + {"Event": {"Attribute": [attribute()]}}, + {"Attribute": [attribute(value="192.0.2.2")]}, + {"Event": {"id": "42"}, "Attribute": [attribute(value="192.0.2.3")]}, + {"Event": {"Object": [{"Attribute": [attribute(value="192.0.2.4")]}]}}, + {"Event": {"id": "43"}, "Object": [{"Attribute": [attribute(value="192.0.2.5")]}]}, + {"Attribute": [attribute()]}, + ] + original = copy.deepcopy(data) + first, result = run(data=data) + assert result["summary"] == {"count": 5, "skipped": {}} + assert run(data=list(reversed(data)))[0] == first + assert run(data=data[0])[1]["summary"]["count"] == 1 + assert data == original + assert re.findall(r"id:(\d+)", first) == [str(number) for number in range(10000, 10005)] + + +@pytest.mark.parametrize( + "kind,value", + [ + ("ip-src", "all"), + ("ip-src", "192.0.2.1\nSecRuleEngine Off"), + ("ip-src", "192.0.2.1/33"), + ("ip-src", "2001:db8::/129"), + ("ip-src", "fe80::1%eth0"), + ("ip-src|port", "192.0.2.1"), + ("ip-src|port", "192.0.2.1|0"), + ("ip-src|port", "192.0.2.1|65536"), + ("ip-src|port", "192.0.2.1|４４３"), + ("domain", 'example.org"'), + ("domain", "example.org\x00"), + ("domain", "example.org\x7f"), + ("domain", "example.org SecRuleEngine Off"), + ("domain", "*.example.org"), + ("domain", "bad..example"), + ("domain", "bad\ud800.example"), + ("domain", "a" * 64 + ".example"), + ("hostname", "192.0.2.1"), + ("hostname|port", "example.org|80|443"), + ("url", "file:///etc/passwd"), + ("url", "https://user:password@example.org/"), + ("url", "https://example.org:/"), + ("url", "https://example.org:0/"), + ("url", "https://example.org:65536/"), + ("url", "https://[fe80::1%25eth0]/"), + ("url", "https://[v1.example]/"), + ("url", "https://example.org/é"), + ("url", "https://example.org/%xx"), + ("url", "https://example.org/\\file"), + ("uri", "relative/path"), + ("uri", "/path#fragment"), + ("uri", "/path\nSecRuleEngine Off"), + ("user-agent", "agent\r\nSecRuleEngine Off"), + ("user-agent", "ü"), + ("user-agent", 123), + ("domain", None), + ("domain", ""), + ], +) +def test_malformed_values_never_emit_partial_rules(kind, value): + text, result = run([attribute(kind, value), attribute()]) + assert text.count("SecRule ") == 1 + assert "@ipMatch 192.0.2.1" in text + assert result["summary"] == {"count": 1, "skipped": {"malformed": 1}} + + +@pytest.mark.parametrize("kind", ["ip-dst", "ip-dst|port", "domain|ip", "port", "sha256", "email-src", "AS"]) +def test_unsupported_types_do_not_become_broader_blocks(kind): + text, result = run([attribute(kind, "example.org|192.0.2.1")]) + assert "SecRule " not in text + assert result["summary"]["skipped"] == {"unsupported": 1} + + +@pytest.mark.parametrize("value", [10000, "20000", 99999, "", None]) +def test_configurable_ids(value): + text, _ = run([attribute()], config={"rule_id_start": value}) + assert f"id:{value or 10000}," in text + + +@pytest.mark.parametrize("value", [True, False, 10000.0, 9999, 100000, "10000,pass", [], {}]) +def test_invalid_rule_id_configuration(value): + result = export.handler(json.dumps({"data": {}, "config": {"rule_id_start": value}})) + assert "error" in result and "data" not in result + + +def test_id_range_exhaustion_and_chained_rule_ids(): + text, result = run([attribute("ip-src|port", "192.0.2.1|443")], config={"rule_id_start": 99999}) + assert result["summary"]["count"] == 1 + assert text.count("id:") == 1 and text.count("SecRule ") == 2 + assert ',chain"\nSecRule REMOTE_PORT' in text + result = export.handler( + json.dumps( + {"data": {"Attribute": [attribute(), attribute(value="192.0.2.2")]}, "config": {"rule_id_start": 99999}} + ) + ) + assert "error" in result and "data" not in result + + +@pytest.mark.parametrize( + "query", + [ + "{", + "null", + "[]", + "{}", + {"data": None}, + {"data": "bad"}, + {"data": [1]}, + {"data": {"Event": []}}, + {"data": {"Attribute": {}}}, + {"data": {"Object": [1]}}, + {"data": {"Object": [{"Attribute": [1]}]}}, + {"data": {"Attribute": [attribute(to_ids="yes")]}}, + {"data": {}, "config": []}, + ], +) +def test_invalid_requests_return_errors_without_output(query): + result = export.handler(query if isinstance(query, str) else json.dumps(query)) + assert "error" in result and "data" not in result + + +def test_empty_events_metadata_and_offline_export(monkeypatch): + import socket + + def no_dns(*args, **kwargs): + pytest.fail("Exporter must not resolve indicators") + + monkeypatch.setattr(socket, "getaddrinfo", no_dns) + monkeypatch.setattr(socket, "gethostbyname", no_dns) + assert export.handler() is False + for data in ([], {}, {"Event": {}}): + assert "SecRule " not in run(data=data)[0] + _, result = run([attribute("domain", "example.org"), attribute("url", "https://example.org/path")]) + assert any("CONNECT" in message for message in result["warnings"]) + assert any("anomaly" in message for message in result["warnings"]) + assert export.introspection()["input"] == sorted(export.SUPPORTED_TYPES) + assert export.introspection()["outputFileExtension"] == "conf" + assert export.version()["module-type"] == ["export"] + assert export.version()["config"] == ["rule_id_start"] + + +class TestCRSHTTP(AsyncHTTPTestCase): + def get_app(self): + self.previous, self.cache = service.MODULES_HANDLERS.copy(), service.ListModules.CACHE + service.MODULES_HANDLERS.clear() + service.MODULES_HANDLERS.update({"owasp_crs_export": export, "type:owasp_crs_export": "export_mod"}) + service.ListModules.CACHE = None + return Application([(r"/modules", service.ListModules), (r"/query", service.QueryModule)]) + + def tearDown(self): + service.MODULES_HANDLERS.clear() + service.MODULES_HANDLERS.update(self.previous) + service.ListModules.CACHE = self.cache + super().tearDown() + + def test_discovery_and_export(self): + response = self.fetch("/modules") + descriptor = next(module for module in json.loads(response.body) if module["name"] == "owasp_crs_export") + assert descriptor["mispattributes"]["outputFileExtension"] == "conf" + response = self.fetch( + "/query", + method="POST", + body=json.dumps( + {"module": "owasp_crs_export", "data": {"Event": {"Attribute": [attribute(), attribute(to_ids=False)]}}} + ), + ) + assert response.code == 200 + result = json.loads(response.body) + assert b"@ipMatch 192.0.2.1" in base64.b64decode(result["data"]) + assert result["summary"] == {"count": 1, "skipped": {"non-IDS": 1}} + + +@pytest.fixture(scope="module") +def modsecurity(): + """Optional native checks: install libmodsecurity3 (or expose it through LD_LIBRARY_PATH).""" + try: + library = ctypes.CDLL(ctypes.util.find_library("modsecurity") or "libmodsecurity.so.3") + except OSError: + pytest.skip("libmodsecurity3 is not installed") + pointer, string, integer = ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int + signatures = { + "msc_init": (pointer, []), + "msc_create_rules_set": (pointer, []), + "msc_rules_add": (integer, [pointer, string, ctypes.POINTER(string)]), + "msc_new_transaction": (pointer, [pointer, pointer, pointer]), + "msc_process_connection": (integer, [pointer, string, integer, string, integer]), + "msc_process_uri": (integer, [pointer, string, string, string]), + "msc_add_request_header": (integer, [pointer, string, string]), + "msc_process_request_headers": (integer, [pointer]), + "msc_intervention": (integer, [pointer, pointer]), + "msc_intervention_cleanup": (None, [pointer]), + "msc_transaction_cleanup": (None, [pointer]), + "msc_rules_cleanup": (integer, [pointer]), + "msc_cleanup": (None, [pointer]), + } + for name, (restype, argtypes) in signatures.items(): + getattr(library, name).restype = restype + getattr(library, name).argtypes = argtypes + return library + + +class Intervention(ctypes.Structure): + _fields_ = [ + ("status", ctypes.c_int), + ("pause", ctypes.c_int), + ("url", ctypes.c_void_p), + ("log", ctypes.c_void_p), + ("disruptive", ctypes.c_int), + ] + + +def native_status( + library, attributes, *, host="example.org", uri="/", client="198.51.100.1", port=12345, agent="test", mode="On" +): + text, _ = run(attributes) + engine, rules = library.msc_init(), library.msc_create_rules_set() + transaction = None + intervention = Intervention(status=200) + try: + error = ctypes.c_char_p() + # Inherited transformations must not alter exact indicator matching. + config = f'SecRuleEngine {mode}\nSecDefaultAction "phase:1,pass,log,t:lowercase"\n' + text + assert library.msc_rules_add(rules, config.encode(), ctypes.byref(error)) >= 0, error.value + transaction = library.msc_new_transaction(engine, rules, None) + library.msc_process_connection(transaction, client.encode(), port, b"203.0.113.1", 80) + library.msc_process_uri(transaction, uri.encode(), b"GET", b"1.1") + library.msc_add_request_header(transaction, b"Host", host.encode()) + library.msc_add_request_header(transaction, b"User-Agent", agent.encode()) + library.msc_process_request_headers(transaction) + library.msc_intervention(transaction, ctypes.byref(intervention)) + return intervention.status + finally: + library.msc_intervention_cleanup(ctypes.byref(intervention)) + if transaction: + library.msc_transaction_cleanup(transaction) + library.msc_rules_cleanup(rules) + library.msc_cleanup(engine) + + +@pytest.mark.parametrize( + "kind,value,traffic,blocked", + [ + ("ip-src", "198.51.100.7/24", {}, True), + ("ip-src", "2001:db8::/32", {"client": "2001:db8::7"}, True), + ("ip-src", "192.0.2.1", {}, False), + ("ip-src|port", "198.51.100.1|12345", {}, True), + ("ip-src|port", "198.51.100.1|12345", {"port": 54321}, False), + ("ip-src|port", "192.0.2.1|12345", {}, False), + ("domain", "EXAMPLE.org.", {"host": "a.EXAMPLE.org.:443"}, True), + ("domain", "example.org", {"host": "evilexample.org"}, False), + ("domain", "example.org", {"host": "example.org.evil.test"}, False), + ("hostname", "example.org", {"host": "a.example.org"}, False), + ("hostname", "bücher.example", {"host": "xn--bcher-kva.example:80"}, True), + ("hostname|port", "EXAMPLE.org|00443", {"host": "example.org:443"}, True), + ("hostname|port", "example.org|443", {"host": "example.org:80"}, False), + ("hostname|port", "example.org|443", {}, False), + ("url", "https://example.org/A?x=1#fragment", {"uri": "/A?x=1"}, True), + ("url", "https://example.org/A?x=1", {"uri": "/a?x=1"}, False), + ("url", "https://example.org/A?x=1", {"uri": "/A?x=1&y=2"}, False), + ("url", "https://example.org/A?x=1", {"host": "other.example", "uri": "/A?x=1"}, False), + ("url", "https://example.org/A?x=1", {"uri": "https://EXAMPLE.org:443/A?x=1"}, True), + ("url", "https://example.org/A?x=1", {"uri": "http://example.org/A?x=1"}, False), + ("url", "https://example.org:8443/", {"host": "example.org:8443"}, True), + ("url", "https://example.org:8443/", {}, False), + ("url", "http://[2001:db8::1]:8080/", {"host": "[2001:db8::1]:8080"}, True), + ("url", "http://example.org", {}, True), + ("url", "https://example.org/A", {"uri": "HTTPS://EXAMPLE.org/A"}, True), + ("url", "http://example.org/?", {"uri": "/?"}, True), + ("url", "http://example.org/?", {}, False), + ("url", "http://example.org/a'b", {"uri": "/a'b"}, True), + ("uri", "/A+(b)?q=$x&y=%20", {"uri": "/A+(b)?q=$x&y=%20"}, True), + ("uri", "/A", {"uri": "/a"}, False), + ("uri", "/A", {"uri": "/AA"}, False), + ("user-agent", "Bot/1.0 (Exact)", {"agent": "Bot/1.0 (Exact)"}, True), + ("user-agent", "Bot/1.0 (Exact)", {"agent": "bot/1.0 (exact)"}, False), + ("user-agent", "x\" ,pass,' \\ %{TX.foo} #", {"agent": "x\" ,pass,' \\ %{TX.foo} #"}, True), + ], +) +def test_native_modsecurity_matching(modsecurity, kind, value, traffic, blocked): + assert native_status(modsecurity, [attribute(kind, value)], **traffic) == (403 if blocked else 200) + + +def test_native_detection_only_and_ids_filter(modsecurity): + assert native_status(modsecurity, [attribute("domain", "example.org")], mode="DetectionOnly") == 200 + assert native_status(modsecurity, [attribute("domain", "example.org", to_ids="false")]) == 200