feat: add TCP probe type

docs/src/components/Layout.jsx

@@ -18,6 +18,7 @@ const navigation = [
             {title: 'Getting started', href: '/'},
             {title: 'Validating HTTP Controls', href: '/docs/http'},
             {title: 'Validating DNS Controls', href: '/docs/dns'},
+            {title: 'Validating TCP Connectivity', href: '/docs/tcp'},
             {title: 'Validating K8s Data', href: '/docs/k8s_data'},
         ],
     },

docs/src/pages/docs/tcp.md

@@ -0,0 +1,132 @@
+---
+title: TCP NetworkAssertions
+description: TCP NetworkAssertions allow you to test raw TCP connectivity from your cluster.
+---
+
+This example shows how to write and run a `NetworkAssertion` that checks TCP connectivity
+from within a namespace. TCP probes verify that a connection can (or cannot) be established
+to a given host and port — the correct primitive for testing connectivity to non-HTTP services
+such as databases, caches, and message brokers.
+
+## TCP NetworkAssertion
+
+We create a `NetworkAssertion` to verify that a TCP connection to the Kubernetes API
+on port 443 succeeds, and that connections to a non-existent service are blocked:
+
+```yaml
+apiVersion: netchecks.io/v1
+kind: NetworkAssertion
+metadata:
+  name: tcp-connectivity
+  namespace: default
+  annotations:
+    description: Assert TCP connectivity to expected services
+spec:
+  schedule: "*/10 * * * *"
+  rules:
+    - name: tcp-to-k8s-api
+      type: tcp
+      host: kubernetes.default.svc
+      port: 443
+      expected: pass
+      validate:
+        message: TCP connection to Kubernetes API should succeed.
+    - name: tcp-to-blocked-port
+      type: tcp
+      host: kubernetes.default.svc
+      port: 9999
+      timeout: 3
+      expected: fail
+      validate:
+        message: TCP connection to non-listening port should fail.
+```
+
+### Parameters
+
+| Parameter | Description | Default |
+| --- | --- | --- |
+| `host` | Hostname or IP address to connect to | (required) |
+| `port` | TCP port number | (required) |
+| `timeout` | Connection timeout in seconds | `5` |
+| `expected` | Whether the check should `pass` or `fail` | `pass` |
+
+## Boundary Protection Example
+
+TCP probes are ideal for verifying network segmentation and boundary protection.
+For example, asserting that a web tier cannot directly reach a database:
+
+```yaml
+apiVersion: netchecks.io/v1
+kind: NetworkAssertion
+metadata:
+  name: boundary-protection
+  namespace: production
+  annotations:
+    description: Verify network segmentation between tiers
+spec:
+  schedule: "@hourly"
+  rules:
+    - name: api-reachable
+      type: tcp
+      host: api.backend
+      port: 8080
+      expected: pass
+      validate:
+        message: Web tier should reach the API tier.
+    - name: database-blocked
+      type: tcp
+      host: postgres.database
+      port: 5432
+      expected: fail
+      validate:
+        message: Web tier must not directly access database tier.
+```
+
+## Custom Validation Rules
+
+You can write custom CEL validation rules to inspect the probe result data:
+
+```yaml
+    - name: tcp-with-custom-rule
+      type: tcp
+      host: my-service.default.svc
+      port: 8080
+      validate:
+        pattern: "data.connected == true && data.error == null"
+        message: TCP connection should succeed with no errors.
+```
+
+The `data` object contains:
+- `connected` (bool) — whether the TCP connection was established
+- `error` (string or null) — error message if the connection failed
+- `startTimestamp` — ISO 8601 timestamp when the check began
+- `endTimestamp` — ISO 8601 timestamp when the check completed
+
+## Policy Report
+
+After the `NetworkAssertion` has been applied, a `PolicyReport` will be created with the
+results. An example `PolicyReport` for a TCP check:
+
+```yaml
+apiVersion: wgpolicyk8s.io/v1alpha2
+kind: PolicyReport
+metadata:
+  name: tcp-connectivity
+  namespace: default
+results:
+  - category: tcp
+    message: Rule from tcp-to-k8s-api
+    policy: tcp-to-k8s-api
+    properties:
+      data: >-
+        {"startTimestamp": "2024-01-15T10:30:00.123456",
+        "connected": true, "error": null,
+        "endTimestamp": "2024-01-15T10:30:00.234567"}
+      spec: >-
+        {"type": "tcp", "host": "kubernetes.default.svc",
+        "port": 443, "timeout": 5}
+    result: pass
+    source: netcheck
+summary:
+  pass: 1
+```

netcheck/checks/tcp.py

@@ -0,0 +1,44 @@
+import datetime
+import logging
+import socket
+
+logger = logging.getLogger("netcheck.tcp")
+DEFAULT_TCP_VALIDATION_RULE = """
+data.connected == true
+"""
+
+
+def tcp_check(host: str, port: int, timeout: float = 5) -> dict:
+    test_spec = {
+        "type": "tcp",
+        "host": host,
+        "port": port,
+        "timeout": timeout,
+    }
+
+    result_data = {
+        "startTimestamp": datetime.datetime.utcnow().isoformat(),
+    }
+
+    output = {"spec": test_spec, "data": result_data}
+
+    try:
+        with socket.create_connection((host, port), timeout=timeout):
+            result_data["connected"] = True
+            result_data["error"] = None
+    except socket.timeout:
+        logger.debug(f"TCP connection to {host}:{port} timed out")
+        result_data["connected"] = False
+        result_data["error"] = f"Connection timed out after {timeout}s"
+    except ConnectionRefusedError:
+        logger.debug(f"TCP connection to {host}:{port} refused")
+        result_data["connected"] = False
+        result_data["error"] = f"Connection refused to {host}:{port}"
+    except OSError as e:
+        logger.debug(f"TCP connection to {host}:{port} failed: {e}")
+        result_data["connected"] = False
+        result_data["error"] = str(e)
+
+    result_data["endTimestamp"] = datetime.datetime.utcnow().isoformat()
+
+    return output

netcheck/cli.py

@@ -8,6 +8,7 @@
 from typing import List, Optional
 
 from netcheck.checks.dns import DEFAULT_DNS_VALIDATION_RULE
+from netcheck.checks.tcp import DEFAULT_TCP_VALIDATION_RULE
 from netcheck.checks.http import NetcheckHttpMethod
 from netcheck.runner import run_from_config, check_individual_assertion
 from netcheck.version import NETCHECK_VERSION
@@ -28,6 +29,7 @@ class NetcheckOutputType(str, Enum):
 class NetcheckTestType(str, Enum):
     dns = "dns"
     http = "http"
+    tcp = "tcp"
     internal = "internal"
 
 
@@ -180,6 +182,45 @@ def dns(
     output_result(result, should_fail, verbose)
 
 
+@app.command()
+def tcp(
+    host: str = typer.Option("github.com", help="Host to connect to", rich_help_panel="tcp test"),
+    port: int = typer.Option(443, help="Port to connect to", rich_help_panel="tcp test"),
+    timeout: float = typer.Option(5.0, "-t", "--timeout", help="Timeout in seconds"),
+    should_fail: bool = typer.Option(False, "--should-fail/--should-pass"),
+    validation_rule: str = typer.Option(None, "--validation-rule", help="Validation rule in CEL to apply to result"),
+    verbose: bool = typer.Option(False, "-v", "--verbose"),
+):
+    """Carry out a tcp connectivity check"""
+
+    test_config = {
+        "host": host,
+        "port": port,
+        "timeout": timeout,
+        "expected": "fail" if should_fail else None,
+    }
+    if verbose:
+        err_console.print("netcheck tcp")
+        err_console.print("Options")
+        err_console.print_json(data=test_config)
+
+    if validation_rule is None:
+        validation_rule = DEFAULT_TCP_VALIDATION_RULE
+    else:
+        err_console.print("Validating result against custom validation rule")
+
+    result = check_individual_assertion(
+        NetcheckTestType.tcp,
+        test_config,
+        err_console,
+        validation_rule=validation_rule,
+        verbose=verbose,
+        include_context=True,
+    )
+
+    output_result(result, should_fail, verbose)
+
+
 def notify_for_unexpected_test_result(failed, should_fail, verbose=False):
     if verbose:
         if failed:

netcheck/runner.py

@@ -11,6 +11,7 @@
 from netcheck.checks.internal import internal_check
 from netcheck.checks.dns import dns_lookup_check, DEFAULT_DNS_VALIDATION_RULE
 from netcheck.checks.http import http_request_check, DEFAULT_HTTP_VALIDATION_RULE
+from netcheck.checks.tcp import tcp_check, DEFAULT_TCP_VALIDATION_RULE
 from netcheck.context import replace_template, LazyFileLoadingDict
 
 logger = logging.getLogger("netcheck.runner")
@@ -108,6 +109,14 @@ def check_individual_assertion(
                 timeout=test_config.get("timeout"),
                 verify=test_config.get("verify-tls-cert", True),
             )
+        case "tcp":
+            if verbose:
+                err_console.print(f"TCP check connecting to {test_config['host']}:{test_config['port']}")
+            test_detail = tcp_check(
+                host=test_config["host"],
+                port=int(test_config["port"]),
+                timeout=test_config.get("timeout", 5),
+            )
         case "internal":
             if verbose:
                 err_console.print(f"Internal check with command '{test_config['command']}'")
@@ -125,6 +134,8 @@ def check_individual_assertion(
                 validation_rule = DEFAULT_HTTP_VALIDATION_RULE
             case "dns":
                 validation_rule = DEFAULT_DNS_VALIDATION_RULE
+            case "tcp":
+                validation_rule = DEFAULT_TCP_VALIDATION_RULE
             case "internal":
                 validation_rule = "true"
             case _:

operator/examples/cilium-tcp-egress-restrictions/tcp-egress-assertion.yaml

@@ -0,0 +1,24 @@
+apiVersion: netchecks.io/v1
+kind: NetworkAssertion
+metadata:
+  name: tcp-egress-restrictions-should-work
+  annotations:
+    description: Verify TCP egress restrictions are enforced by Cilium network policy
+spec:
+  schedule: "*/1 * * * *"
+  rules:
+    - name: tcp-to-k8s-api-allowed
+      type: tcp
+      host: kubernetes.default.svc
+      port: 443
+      expected: pass
+      validate:
+        message: TCP to Kubernetes API port 443 should be allowed by policy.
+    - name: tcp-to-blocked-port
+      type: tcp
+      host: kubernetes.default.svc
+      port: 8080
+      timeout: 5
+      expected: fail
+      validate:
+        message: TCP to port 8080 should be blocked by Cilium egress policy.

operator/examples/cilium-tcp-egress-restrictions/tcp-egress-netpol.yaml

@@ -0,0 +1,28 @@
+apiVersion: "cilium.io/v2"
+kind: CiliumNetworkPolicy
+metadata:
+  name: restrict-tcp-egress
+spec:
+  endpointSelector: {}
+  egress:
+    # Allow DNS to kube-dns (required for name resolution)
+    - toEndpoints:
+        - matchLabels:
+            "k8s:io.kubernetes.pod.namespace": kube-system
+            "k8s:k8s-app": kube-dns
+      toPorts:
+        - ports:
+            - port: "53"
+              protocol: ANY
+          rules:
+            dns:
+              - matchPattern: "*"
+    # Allow TCP to the Kubernetes API (port 443)
+    - toServices:
+        - k8sService:
+            serviceName: kubernetes
+            namespace: default
+      toPorts:
+        - ports:
+            - port: "443"
+              protocol: TCP

operator/examples/compliance/soc2-boundary-protection.yaml

@@ -0,0 +1,30 @@
+apiVersion: netchecks.io/v1
+kind: NetworkAssertion
+metadata:
+  name: boundary-protection-web-tier
+  namespace: production
+  annotations:
+    netchecks.io/controls: "soc2/CC6.6, soc2/CC6.7"
+    netchecks.io/description: >
+      Verify web tier boundary protection is effective. Web pods
+      should reach the API tier but not the database tier directly.
+      SOC 2 CC6.6 requires boundary protection mechanisms to be
+      in place and monitored.
+    netchecks.io/severity: high
+spec:
+  schedule: "@hourly"
+  rules:
+    - name: web-to-api-allowed
+      type: tcp
+      host: api.backend
+      port: 8080
+      expected: pass
+      validate:
+        message: Web tier should reach the API tier
+    - name: web-to-db-blocked
+      type: tcp
+      host: postgres.database
+      port: 5432
+      expected: fail
+      validate:
+        message: Web tier must not directly access database tier

operator/examples/default-k8s/tcp.yaml

@@ -0,0 +1,25 @@
+apiVersion: netchecks.io/v1
+kind: NetworkAssertion
+metadata:
+  name: tcp-should-work
+  namespace: default
+  annotations:
+    description: Assert pod can establish TCP connections to expected services.
+spec:
+  schedule: "*/10 * * * *"
+  rules:
+    - name: tcp-to-k8s-api
+      type: tcp
+      host: kubernetes.default.svc
+      port: 443
+      expected: pass
+      validate:
+        message: TCP connection to Kubernetes API should succeed.
+    - name: tcp-to-blocked-port
+      type: tcp
+      host: kubernetes.default.svc
+      port: 9999
+      timeout: 3
+      expected: fail
+      validate:
+        message: TCP connection to non-listening port should fail.