Audit HTTP Cookie Domain Isolation with the Mozilla Public Suffix List in DuckDB

id: mozilla-public-suffix-cookie-isolation
namespace: company.security

description: |
  Audit application endpoints and API gateways against the official Mozilla Public
  Suffix List to detect supercookie domain leaks, unencrypted transport, and RFC 6265
  isolation violations using cached downloads and DuckDB SQL.

inputs:
  - id: target_endpoints
    type: ARRAY
    itemType: STRING
    description: Web application or API gateway URLs to probe for Set-Cookie header
      compliance.
    defaults:
      - https://kestra.io
      - https://github.com

  - id: dry_run
    type: BOOL
    description: When true, audits cookies and exports reports without failing the
      execution on security violations.
    defaults: true

  - id: fail_on_supercookie
    type: BOOL
    description: When true and dry_run is false, fails the pipeline immediately upon
      detecting a supercookie.
    defaults: false

triggers:
  - id: weekly_security_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run the cookie boundary and public suffix audit every Monday at 04:00 UTC.
    cron: 0 4 * * 1
    disabled: true

tasks:
  - id: download_mozilla_psl
    type: io.kestra.plugin.core.http.Download
    description: Fetch the official Mozilla Public Suffix List with a 24-hour task
      cache to minimize upstream load.
    uri: https://publicsuffix.org/list/public_suffix_list.dat
    taskCache:
      enabled: true
      ttl: PT24H

  - id: probe_cookie_endpoints
    type: io.kestra.plugin.scripts.python.Script
    description: Probe configured endpoints via Python stdlib to capture Set-Cookie
      attributes into an audit table.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    outputFiles:
      - observed_cookies.csv
    script: |
      import csv
      import json
      import sys
      import urllib.request
      import urllib.parse
      from http.cookies import SimpleCookie

      endpoints = json.loads("""{{ inputs.target_endpoints | toJson }}""")
      output_path = "observed_cookies.csv"

      rows = []
      headers = ["endpoint", "cookie_name", "domain_scope", "path", "is_secure", "is_httponly", "same_site"]

      for url in endpoints:
          try:
              req = urllib.request.Request(
                  url,
                  headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Kestra-Security-Audit/1.0"}
              )
              with urllib.request.urlopen(req, timeout=15) as resp:
                  raw_cookies = resp.headers.get_all("Set-Cookie") or []
                  parsed_host = urllib.parse.urlparse(url).hostname or ""
                  for raw in raw_cookies:
                      cookie = SimpleCookie()
                      try:
                          cookie.load(raw)
                          for key, morsel in cookie.items():
                              domain = morsel["domain"] if morsel["domain"] else parsed_host
                              rows.append({
                                  "endpoint": url,
                                  "cookie_name": key,
                                  "domain_scope": domain,
                                  "path": morsel["path"] or "/",
                                  "is_secure": "true" if morsel["secure"] else "false",
                                  "is_httponly": "true" if morsel["httponly"] else "false",
                                  "same_site": morsel["samesite"] or "None"
                              })
                      except Exception as parse_err:
                          print(f"Warning: Could not parse cookie string '{raw}': {parse_err}", file=sys.stderr)
          except Exception as req_err:
              print(f"Warning: Failed to probe endpoint '{url}': {req_err}", file=sys.stderr)

      # Ensure observed_cookies.csv always exists with standard schema even if zero cookies were set
      with open(output_path, "w", newline="", encoding="utf-8") as f:
          writer = csv.DictWriter(f, fieldnames=headers)
          writer.writeheader()
          for r in rows:
              writer.writerow(r)

      print(f"Captured {len(rows)} cookies across {len(endpoints)} endpoints.")

  - id: duckdb_psl_analysis
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: Parse the Mozilla PSL rules, cross-reference cookie scopes, and
      export a Parquet audit report.
    inputFiles:
      psl.dat: "{{ outputs.download_mozilla_psl.uri }}"
      observed_cookies.csv: "{{
        outputs.probe_cookie_endpoints.outputFiles['observed_cookies.csv'] }}"
    outputFiles:
      - cookie_isolation_audit.parquet
    sql: |
      CREATE TABLE raw_lines AS 
      SELECT 
        row_number() OVER () AS line_number, 
        line 
      FROM read_csv('{{ workingDir }}/psl.dat', header=false, delim='\n', quote='', escape='', columns={'line': 'VARCHAR'});

      CREATE TABLE psl_rules AS
      WITH labeled AS (
        SELECT 
          line_number,
          trim(line) AS rule,
          CASE 
            WHEN line LIKE '%===BEGIN ICANN DOMAINS===' THEN 'ICANN'
            WHEN line LIKE '%===BEGIN PRIVATE DOMAINS===' THEN 'PRIVATE'
            ELSE NULL
          END AS marker
        FROM raw_lines
      ),
      sectioned AS (
        SELECT 
          line_number,
          rule,
          last_value(marker IGNORE NULLS) OVER (ORDER BY line_number ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS section
        FROM labeled
      )
      SELECT 
        CASE 
          WHEN rule LIKE '!%' THEN lower(trim(substr(rule, 2)))
          WHEN rule LIKE '*.%' THEN lower(trim(substr(rule, 3)))
          ELSE lower(trim(rule))
        END AS clean_suffix,
        lower(trim(rule)) AS raw_rule,
        section,
        CASE 
          WHEN rule LIKE '!%' THEN 'EXCEPTION'
          WHEN rule LIKE '*.%' THEN 'WILDCARD'
          ELSE 'EXACT'
        END AS rule_type
      FROM sectioned
      WHERE rule != '' 
        AND rule NOT LIKE '//%' 
        AND rule NOT LIKE '%===%';

      CREATE TABLE observed AS
      SELECT 
        endpoint,
        cookie_name,
        domain_scope,
        path,
        CAST(is_secure AS BOOLEAN) AS is_secure,
        CAST(is_httponly AS BOOLEAN) AS is_httponly,
        same_site,
        lower(ltrim(domain_scope, '.')) AS clean_domain
      FROM read_csv_auto('{{ workingDir }}/observed_cookies.csv', header=True);

      CREATE TABLE candidate_matches AS
      SELECT 
        o.endpoint,
        o.cookie_name,
        o.domain_scope,
        o.clean_domain,
        o.path,
        o.is_secure,
        o.is_httponly,
        o.same_site,
        p.clean_suffix,
        p.raw_rule,
        p.section,
        p.rule_type,
        CASE 
          WHEN p.rule_type = 'EXCEPTION' AND o.clean_domain = p.clean_suffix THEN 1
          WHEN p.rule_type = 'EXACT' AND o.clean_domain = p.clean_suffix THEN 2
          WHEN p.rule_type = 'WILDCARD' AND (
            o.clean_domain = p.clean_suffix 
            OR (o.clean_domain LIKE '%.' || p.clean_suffix AND length(o.clean_domain) - length(replace(o.clean_domain, '.', '')) = length(p.clean_suffix) - length(replace(p.clean_suffix, '.', '')) + 1)
          ) THEN 3
          ELSE 99
        END AS match_priority
      FROM observed o
      LEFT JOIN psl_rules p ON 
        o.clean_domain = p.clean_suffix
        OR (p.rule_type = 'WILDCARD' AND (
          o.clean_domain = p.clean_suffix
          OR o.clean_domain LIKE '%.' || p.clean_suffix
        ));

      CREATE TABLE cookie_audit AS
      WITH ranked_matches AS (
        SELECT *,
          row_number() OVER (PARTITION BY endpoint, cookie_name, clean_domain ORDER BY match_priority ASC) AS rank
        FROM candidate_matches
        WHERE match_priority < 99
      )
      SELECT 
        o.endpoint,
        o.cookie_name,
        o.domain_scope,
        o.clean_domain,
        r.section AS psl_section,
        r.rule_type AS psl_rule_type,
        r.raw_rule AS matched_psl_rule,
        CASE 
          WHEN r.rule_type = 'EXCEPTION' THEN 'COMPLIANT'
          WHEN r.clean_suffix IS NOT NULL THEN 'CRITICAL_SUPERCOOKIE'
          WHEN NOT o.is_secure THEN 'HIGH_INSECURE_TRANSPORT'
          WHEN NOT o.is_httponly THEN 'MEDIUM_MISSING_HTTPONLY'
          WHEN o.same_site IS NULL OR lower(o.same_site) = 'none' OR trim(o.same_site) = '' THEN 'MEDIUM_SAMESITE_CROSS_SITE'
          ELSE 'COMPLIANT'
        END AS security_verdict,
        CASE 
          WHEN r.rule_type = 'EXCEPTION' THEN 'Cookie domain explicitly exempt from wildcard public suffix rule.'
          WHEN r.clean_suffix IS NOT NULL THEN 'RFC 6265 violation: Cookie scoped to public suffix; exposes credentials across all tenant subdomains.'
          WHEN NOT o.is_secure THEN 'Cookie transmitted over unencrypted HTTP or lacks Secure attribute.'
          WHEN NOT o.is_httponly THEN 'Cookie accessible via client-side JavaScript document.cookie (XSS risk).'
          WHEN o.same_site IS NULL OR lower(o.same_site) = 'none' OR trim(o.same_site) = '' THEN 'Cookie lacks Lax or Strict SameSite isolation (CSRF exposure).'
          ELSE 'Cookie scope and security flags adhere to RFC 6265bis security baseline.'
        END AS risk_rationale
      FROM observed o
      LEFT JOIN (SELECT * FROM ranked_matches WHERE rank = 1) r 
        ON o.endpoint = r.endpoint 
        AND o.cookie_name = r.cookie_name 
        AND o.clean_domain = r.clean_domain;

      COPY cookie_audit TO '{{ workingDir }}/cookie_isolation_audit.parquet' (FORMAT PARQUET, COMPRESSION ZSTD);

      SELECT 
        count(*) AS total_cookies_audited,
        count(*) FILTER (WHERE security_verdict = 'CRITICAL_SUPERCOOKIE') AS critical_supercookies,
        count(*) FILTER (WHERE security_verdict = 'HIGH_INSECURE_TRANSPORT') AS insecure_transport,
        count(*) FILTER (WHERE security_verdict = 'MEDIUM_MISSING_HTTPONLY') AS missing_httponly,
        count(*) FILTER (WHERE security_verdict = 'COMPLIANT') AS compliant_cookies
      FROM cookie_audit;
    fetchType: FETCH_ONE

  - id: check_security_violations
    type: io.kestra.plugin.core.flow.If
    description: Branch based on whether any supercookie leaks or insecure transport
      attributes were detected.
    condition: "{{ outputs.duckdb_psl_analysis.outputs[0].row.critical_supercookies
      > 0 or outputs.duckdb_psl_analysis.outputs[0].row.insecure_transport > 0
      }}"
    then:
      - id: notify_security_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alert the security engineering channel with observed cookie
          boundary violations.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#appsec-alerts",
            "text": ":warning: *Cookie Isolation Security Alert*\nAudited: `{{ outputs.duckdb_psl_analysis.outputs[0].row.total_cookies_audited }}` cookies\nCritical Supercookies: `{{ outputs.duckdb_psl_analysis.outputs[0].row.critical_supercookies }}`\nInsecure Transport: `{{ outputs.duckdb_psl_analysis.outputs[0].row.insecure_transport }}`\nMissing HttpOnly: `{{ outputs.duckdb_psl_analysis.outputs[0].row.missing_httponly }}`\nParquet Audit: `{{ outputs.duckdb_psl_analysis.outputFiles['cookie_isolation_audit.parquet'] }}`"
          }

      - id: log_security_breach
        type: io.kestra.plugin.core.log.Log
        description: Log warning metrics for observability dashboards when isolation
          boundaries fail.
        message: "Security warning: Found {{
          outputs.duckdb_psl_analysis.outputs[0].row.critical_supercookies }}
          supercookies and {{
          outputs.duckdb_psl_analysis.outputs[0].row.insecure_transport }}
          unencrypted cookies across probed endpoints."
    else:
      - id: log_clean_audit
        type: io.kestra.plugin.core.log.Log
        description: Log success confirmation when all audited cookies observe Public
          Suffix boundaries.
        message: "Security audit clean: All {{
          outputs.duckdb_psl_analysis.outputs[0].row.total_cookies_audited }}
          cookies observe RFC 6265bis public suffix isolation rules."

  - id: enforce_isolation_gate
    type: io.kestra.plugin.core.execution.Assert
    description: Block pipeline progression if critical supercookies are discovered
      and enforcement is active.
    conditions:
      - "{{ inputs.dry_run == true or inputs.fail_on_supercookie == false or
        outputs.duckdb_psl_analysis.outputs[0].row.critical_supercookies == 0
        }}"
    errorMessage: "Critical RFC 6265 security violation: One or more cookies are
      scoped to a public suffix, enabling cross-tenant session hijacking."