feat: support custom messages

In some cases you might need some additional messages. By adding a additional message to the definition it will be rendered in the rule.
binxio · Jul 28, 2023 · 861ce24 · 861ce24
1 parent bab935f
commit 861ce24
Show file tree

Hide file tree

Showing 9 changed files with 74 additions and 10 deletions.
diff --git a/aws_network_firewall/destination.py b/aws_network_firewall/destination.py
@@ -17,6 +17,7 @@ class Destination:
     endpoint: Optional[str]
     region: Optional[str]
     cidr: Optional[str]
+    message: Optional[str]
 
     def resolve_region_cidr_ranges(self, ranges: CidrRanges) -> None:
         if self.region and not self.cidr:

diff --git a/aws_network_firewall/rule.py b/aws_network_firewall/rule.py
@@ -72,8 +72,14 @@ def __resolve_options(self, destination: Destination) -> List[SuricataOption]:
         if destination.protocol == "TLS" and destination.endpoint:
             options = self.__tls_endpoint_options(destination.endpoint)
 
+        message = (
+            f"{destination.message} | {self.workload} | {self.name}"
+            if destination.message
+            else f"{self.workload} | {self.name}"
+        )
+
         return options + [
-            SuricataOption(name="msg", value=f"{self.workload} | {self.name}"),
+            SuricataOption(name="msg", value=message),
             SuricataOption(name="rev", value="1", quoted_value=False),
             SuricataOption(name="sid", value="XXX", quoted_value=False),
         ]

diff --git a/aws_network_firewall/schemas/__init__.py b/aws_network_firewall/schemas/__init__.py
@@ -28,6 +28,7 @@ def destination_resolver(entry: dict) -> Destination:
         endpoint=entry.get("Endpoint"),
         region=entry.get("Region"),
         cidr=entry.get("Cidr"),
+        message=entry.get("Message"),
     )
 
 

diff --git a/aws_network_firewall/schemas/environment.yaml b/aws_network_firewall/schemas/environment.yaml
@@ -113,6 +113,8 @@ definitions:
         enum: [ "TCP", "TLS", "ICMP" ]
       Port:
         type: integer
+      Message:
+        type: string
     examples:
       - Description: Website of Xebia
         Protocol: TLS

diff --git a/aws_network_firewall/suricata/option.py b/aws_network_firewall/suricata/option.py
@@ -20,4 +20,4 @@ def __str__(self):
         if self.quoted_value:
             value = f'"{self.value}"'
 
-        return self.name if not self.value else f"{self.name}: {value}"
+        return self.name if not self.value else f"{self.name}:{value}"
diff --git a/tests/test_account.py b/tests/test_account.py
@@ -21,6 +21,7 @@ def generate_rule(type: str) -> Rule:
                 cidr=None,
                 endpoint=None,
                 region=None,
+                message=None,
             )
         ],
     )

diff --git a/tests/test_destination.py b/tests/test_destination.py
@@ -10,6 +10,7 @@ def test_destination_region_to_cidr(cidr_ranges: CidrRanges) -> None:
         region="eu-west-1",
         endpoint=None,
         cidr=None,
+        message=None,
     )
     destination.resolve_region_cidr_ranges(cidr_ranges)
     assert destination.cidr == "10.0.0.0/24"
@@ -23,6 +24,26 @@ def test_destination_cidr(cidr_ranges: CidrRanges) -> None:
         region=None,
         endpoint=None,
         cidr="10.0.0.0/24",
+        message=None,
     )
     destination.resolve_region_cidr_ranges(cidr_ranges)
     assert destination.cidr == "10.0.0.0/24"
+
+
+def test_destination_properties() -> None:
+    destination = Destination(
+        description="My Description",
+        protocol="TLS",
+        port=443,
+        region="eu-west-1",
+        endpoint="xebia.com",
+        cidr="10.0.0.0/24",
+        message="Important Message",
+    )
+    assert destination.description == "My Description"
+    assert destination.protocol == "TLS"
+    assert destination.port == 443
+    assert destination.region == "eu-west-1"
+    assert destination.endpoint == "xebia.com"
+    assert destination.cidr == "10.0.0.0/24"
+    assert destination.message == "Important Message"
diff --git a/tests/test_rule.py b/tests/test_rule.py
@@ -18,12 +18,13 @@ def test_rule_with_tls_endpoint() -> None:
                 endpoint="xebia.com",
                 cidr="10.0.1.0/24",
                 region=None,
+                message=None,
             )
         ],
     )
 
     assert (
-        'pass tls 10.0.0.0/24 any -> 10.0.1.0/24 443 (tls.sni; tls.version: 1.2; content: "xebia.com"; nocase; startswith; endswith; msg: "my-workload | my-rule"; rev: 1; sid: XXX;)'
+        'pass tls 10.0.0.0/24 any -> 10.0.1.0/24 443 (tls.sni; tls.version:1.2; content:"xebia.com"; nocase; startswith; endswith; msg:"my-workload | my-rule"; rev:1; sid:XXX;)'
         == str(rule)
     )
 
@@ -43,12 +44,13 @@ def test_rule_with_tls_wildcard_endpoint() -> None:
                 endpoint="*.xebia.com",
                 cidr="10.0.1.0/24",
                 region=None,
+                message=None,
             )
         ],
     )
 
     assert (
-        'pass tls 10.0.0.0/24 any -> 10.0.1.0/24 443 (tls.sni; tls.version: 1.2; dotprefix; content: ".xebia.com"; nocase; endswith; msg: "my-workload | my-rule"; rev: 1; sid: XXX;)'
+        'pass tls 10.0.0.0/24 any -> 10.0.1.0/24 443 (tls.sni; tls.version:1.2; dotprefix; content:".xebia.com"; nocase; endswith; msg:"my-workload | my-rule"; rev:1; sid:XXX;)'
         == str(rule)
     )
 
@@ -68,13 +70,14 @@ def test_rule_with_tls_endpoint_non_standard_port() -> None:
                 endpoint="xebia.com",
                 cidr="10.0.1.0/24",
                 region=None,
+                message=None,
             )
         ],
     )
 
     assert (
-        'pass tls 10.0.0.0/24 any -> 10.0.1.0/24 444 (tls.sni; tls.version: 1.2; content: "xebia.com"; nocase; startswith; endswith; msg: "my-workload | my-rule"; rev: 1; sid: XXX;)\n'
-        + 'pass tcp 10.0.0.0/24 any <> 10.0.1.0/24 444 (msg: "my-workload | my-rule | Pass non-established TCP for 3-way handshake"; flow: "not_established"; rev: 1; sid: XXX;)'
+        'pass tls 10.0.0.0/24 any -> 10.0.1.0/24 444 (tls.sni; tls.version:1.2; content:"xebia.com"; nocase; startswith; endswith; msg:"my-workload | my-rule"; rev:1; sid:XXX;)\n'
+        + 'pass tcp 10.0.0.0/24 any <> 10.0.1.0/24 444 (msg:"my-workload | my-rule | Pass non-established TCP for 3-way handshake"; flow:"not_established"; rev:1; sid:XXX;)'
         == str(rule)
     )
 
@@ -94,12 +97,13 @@ def test_rule_with_tcp_cidr() -> None:
                 cidr="10.0.1.0/24",
                 endpoint=None,
                 region=None,
+                message=None,
             )
         ],
     )
 
     assert (
-        'pass tcp 10.0.0.0/24 any -> 10.0.1.0/24 443 (msg: "my-workload | my-rule"; rev: 1; sid: XXX;)'
+        'pass tcp 10.0.0.0/24 any -> 10.0.1.0/24 443 (msg:"my-workload | my-rule"; rev:1; sid:XXX;)'
         == str(rule)
     )
 
@@ -119,12 +123,13 @@ def test_icmp_rule() -> None:
                 cidr="10.0.1.0/24",
                 endpoint=None,
                 region=None,
+                message=None,
             )
         ],
     )
 
     assert (
-        'pass icmp 10.0.0.0/24 any <> 10.0.1.0/24 any (msg: "my-workload | my-rule"; rev: 1; sid: XXX;)'
+        'pass icmp 10.0.0.0/24 any <> 10.0.1.0/24 any (msg:"my-workload | my-rule"; rev:1; sid:XXX;)'
         == str(rule)
     )
 
@@ -144,11 +149,38 @@ def test_egress_tls_rule() -> None:
                 cidr=None,
                 endpoint="xebia.com",
                 region=None,
+                message=None,
             )
         ],
     )
 
     assert (
-        'pass tls  any -> any 443 (tls.sni; tls.version: 1.2; content: "xebia.com"; nocase; startswith; endswith; msg: "my-workload | my-rule"; rev: 1; sid: XXX;)'
+        'pass tls  any -> any 443 (tls.sni; tls.version:1.2; content:"xebia.com"; nocase; startswith; endswith; msg:"my-workload | my-rule"; rev:1; sid:XXX;)'
+        == str(rule)
+    )
+
+
+def test_egress_tls_rule_with_message() -> None:
+    rule = Rule(
+        workload="my-workload",
+        name="my-rule",
+        type=Rule.EGRESS,
+        description="My description",
+        sources=[Source(description="my source", cidr=None, region="eu-west-1")],
+        destinations=[
+            Destination(
+                description="my destination",
+                protocol="TLS",
+                port=443,
+                cidr=None,
+                endpoint="xebia.com",
+                region=None,
+                message="IMPORTANT BECAUSE ...",
+            )
+        ],
+    )
+
+    assert (
+        'pass tls  any -> any 443 (tls.sni; tls.version:1.2; content:"xebia.com"; nocase; startswith; endswith; msg:"IMPORTANT BECAUSE ... | my-workload | my-rule"; rev:1; sid:XXX;)'
         == str(rule)
     )
diff --git a/tests/workloads/example-workload/README.md b/tests/workloads/example-workload/README.md
@@ -50,7 +50,7 @@ xebia.com | 192.168.8.0/21 | eu-central-1 | TLS  | 443 | My destination
 Based on the above defined sources and destination the following firewall rules are required:
 
 ```
-pass tls 192.168.0.0/21 any -> 192.168.8.0/21 443 (tls.sni; tls.version: 1.2; content: "xebia.com"; nocase; startswith; endswith; msg: "binxio-example-workload-development | My Rule name"; rev: 1; sid: XXX;)
+pass tls 192.168.0.0/21 any -> 192.168.8.0/21 443 (tls.sni; tls.version:1.2; content:"xebia.com"; nocase; startswith; endswith; msg:"binxio-example-workload-development | My Rule name"; rev:1; sid:XXX;)
 
 ```