From f410ee8f44b4a4762147e5101801af5152adee3a Mon Sep 17 00:00:00 2001 From: Said Boudjelda Date: Sat, 17 Oct 2026 19:28:06 +0200 Subject: [PATCH] fix(security): the audit and pentest read each listener's protocol, so Kafka's default GSSAPI and named listeners no longer pass SASL Authentication and SASL Mechanism Strength passed when sasl.enabled.mechanisms was set, and Kafka reports GSSAPI for that key on every broker that leaves it alone: a broker with only PLAINTEXT://:9092, SecurityApiIT's container among them, read "SASL mechanisms: GSSAPI", and the pentest called its metadata PROTECTED. On the kafka-cluster chart's Strimzi cluster it went the other way: the broker-wide key is empty, so SASL Authentication failed although listener PLAIN-9092 is SCRAM, and No Plaintext Listeners and the unencrypted probe passed because they looked for the text PLAINTEXT://, which PLAIN-9092 (SASL_PLAINTEXT, no TLS) never holds. Listener Protocol Map passed whenever the map held SASL_PLAINTEXT, and Kafka's default map holds all four protocols. SecurityKafkaHelpers now works out each listener's protocol from listeners and listener.security.protocol.map, a missing listeners counting as Kafka's default PLAINTEXT://:9092. SASL Authentication and metadata-leak pass when some listener is SASL_* and none is PLAINTEXT. No Plaintext Listeners, Listener Protocol Map and the unencrypted probe flag PLAINTEXT and SASL_PLAINTEXT listeners in use, and name them; advertised.listeners is no longer read. SASL Mechanism Strength judges the SASL listeners' listener.name..sasl.enabled.mechanisms, falling back to the broker-wide key, and says "No SASL listener" when there is none. Whether DescribeConfigs returns the per-listener key is not checked on a live cluster yet. SecurityListenerProtocolsTest feeds the audit and the pentest the chart's Strimzi config (with an empty and a GSSAPI broker-wide key), a vanilla PLAINTEXT broker, a SASL_SSL-only broker and a PLAINTEXT listener beside a SASL one. SecurityApiIT now asserts the container fails SASL Authentication, warns on plaintext naming its BROKER listener (which only a config read from the broker holds) and is VULNERABLE to both probes. Caveat security-broker-wide-keys is rewritten to say what the checks read now and that the ssl.* checks are still broker-wide; the skill's bullet follows it. The security refs the edit moved are re-pointed and anchored. Grades change with no config change: the chart's cluster now passes SASL Authentication and warns on No Plaintext Listeners and Listener Protocol Map, and a vanilla broker fails SASL Authentication, which a security gate or a saved drift baseline will show. --- cli/cmd/mcp_caveats.go | 11 +- cli/cmd/mcp_tools_security.go | 61 +++--- eval/mcp/oracle.py | 2 +- .../kates/service/SecurityKafkaHelpers.java | 89 +++++++++ .../kates/service/SecurityPentestService.java | 14 +- .../kates/service/SecurityService.java | 58 +++--- .../com/bmscomp/kates/it/SecurityApiIT.java | 20 ++ .../SecurityListenerProtocolsTest.java | 184 ++++++++++++++++++ skills/kates-cli/SKILL.md | 2 +- 9 files changed, 378 insertions(+), 63 deletions(-) create mode 100644 kates/src/test/java/com/bmscomp/kates/service/SecurityListenerProtocolsTest.java diff --git a/cli/cmd/mcp_caveats.go b/cli/cmd/mcp_caveats.go index 5d5ecebd..b90c09b8 100644 --- a/cli/cmd/mcp_caveats.go +++ b/cli/cmd/mcp_caveats.go @@ -126,7 +126,8 @@ var mcpCaveatsCore = []mcpCaveat{ ID: mcpCaveatPentestConfigOnly, Text: "The security \"pentest\" attacks nothing. Its six checks read the configuration of the first broker " + "listed and the ACL list; an ACL list it cannot read is reported as PROTECTED.", - Refs: []string{mcpJava + "service/SecurityPentestService.java:40-67,102-204"}, + Refs: []string{mcpAnchoredRef(mcpJava+"service/SecurityPentestService.java:40-67,102-206", + "public Map pentest(String testName)", "private Map pentestResult(")}, }, { ID: mcpCaveatCVEFixedList, @@ -134,8 +135,8 @@ var mcpCaveatsCore = []mcpCaveat{ "learns the Kafka version: it reads kafkaVersion from the cluster description, which does not carry one, " + "so the version is \"unknown\", the comparison fails, and every CVE is reported PATCHED with grade PASS.", Refs: []string{ - mcpJava + "service/SecurityService.java:1383-1474", - mcpJava + "service/SecurityService.java:1572-1586", + mcpAnchoredRef(mcpJava+"service/SecurityService.java:1387-1478", "public Map cveCheck()"), + mcpAnchoredRef(mcpJava+"service/SecurityService.java:1576-1590", "private int compareVersions(String v1, String v2)"), mcpJava + "service/ClusterHealthService.java:72-93", }, }, @@ -147,8 +148,8 @@ var mcpCaveatsCore = []mcpCaveat{ "too, so there it mostly reflects API traffic).", Refs: []string{ mcpJava + "service/SecurityService.java:42,58-69", - mcpJava + "service/SecurityService.java:789,852,914,979", - mcpJava + "service/SecurityService.java:1649-1689", + mcpAnchoredRef(mcpJava+"service/SecurityService.java:793,856,918,983", "= runAudit();"), + mcpAnchoredRef(mcpJava+"service/SecurityService.java:1653-1693", "private void recordSnapshot("), }, }, { diff --git a/cli/cmd/mcp_tools_security.go b/cli/cmd/mcp_tools_security.go index 5d8fd53d..f3d9844a 100644 --- a/cli/cmd/mcp_tools_security.go +++ b/cli/cmd/mcp_tools_security.go @@ -68,7 +68,7 @@ var mcpSecuritySections = []string{ } // mcpComplianceFramework names a framework as the tool takes it and as the -// backend keys it (SecurityService.java:796-798). +// backend keys it (SecurityService.java:800-802). type mcpComplianceFrameworkName struct{ key, name string } var mcpComplianceFrameworks = []mcpComplianceFrameworkName{ @@ -710,7 +710,7 @@ const ( // version. It reads kafkaVersion from a cluster description that never holds // one (ClusterHealthService.describeCluster), so it gets "unknown", and its // compareVersions, failing to parse that, calls every entry PATCHED -// (SecurityService.java:1390,1448,1572-1586). A version counts as known when +// (SecurityService.java:1394,1452,1576-1590). A version counts as known when // every dot-separated part holds a digit, as compareVersions needs. func mcpCVEVersionKnown(v string) bool { if v == "" { @@ -936,8 +936,10 @@ var mcpCaveatsSecurity = mcpJoinCaveats([]mcpCaveat{ "backend logs a warning and runs the checks on an empty configuration or an empty ACL list, and the " + "result does not say so: each check then reports the value it assumes for a missing setting.", Refs: []string{ - mcpJava + "service/SecurityKafkaHelpers.java:31-60", - mcpJava + "service/SecurityService.java:88-97,1055-1061,1293-1297,1511-1519", + mcpAnchoredRef(mcpJava+"service/SecurityKafkaHelpers.java:36-65", + "fetchBrokerConfig(AdminClient client, int brokerId)", "fetchAcls(AdminClient client)"), + mcpAnchoredRef(mcpJava+"service/SecurityService.java:88-97,1059-1065,1297-1301,1515-1523", + "helpers.fetchBrokerConfig(client, brokerId)", "helpers.fetchBrokerConfig(client, bId)"), mcpJava + "service/SecurityPentestService.java:46-51", }, }, @@ -950,28 +952,39 @@ var mcpCaveatsSecurity = mcpJoinCaveats([]mcpCaveat{ "pass, and no certificate is opened: expiry, issuer and chain are never read, only whether a keystore and " + "a truststore location are set.", Refs: []string{ - mcpJava + "service/SecurityService.java:1049-1158", - mcpJava + "service/SecurityService.java:1286-1380", + mcpAnchoredRef(mcpJava+"service/SecurityService.java:1053-1162", "public Map tlsInspect()"), + mcpAnchoredRef(mcpJava+"service/SecurityService.java:1290-1384", "public Map certificateCheck()"), }, }, { ID: mcpCaveatSecurityBrokerWideKeys, - Text: "The audit, pentest, TLS and certificate checks read only broker-wide settings (sasl.enabled.mechanisms, " + - "ssl.keystore.location, ssl.truststore.location, ssl.client.auth, ssl.protocol and the like), never " + - "listener.name..* settings. The audit's No Plaintext Listeners check and the pentest's " + - "unencrypted check only look for the text PLAINTEXT:// in the listeners setting (the audit also in " + - "advertised.listeners): a listener named after its protocol (PLAINTEXT, SASL_PLAINTEXT) holds it, and one " + - "with a name of its own, such as plain, never does, whatever its protocol. On a " + - "cluster that sets security per named listener, such as the Strimzi cluster the kafka-cluster chart " + - "installs (listener plain on 9092, SCRAM without TLS; listener tls on 9093, TLS), these checks can report " + - "no SASL, no keystore and TLS unavailable, and at the same time no plaintext listener: read neither " + - "result as the truth about the listeners. The Listener Protocol Map check's detail lists each listener's " + - "protocol unless the map holds PLAINTEXT without SASL_PLAINTEXT; it passes whenever the map holds " + - "SASL_PLAINTEXT, which is authenticated but unencrypted, even beside a PLAINTEXT listener.", + Text: "The audit's SASL Authentication, SASL Mechanism Strength, No Plaintext Listeners and Listener Protocol " + + "Map checks and the pentest's metadata-leak and unencrypted checks take each listener's protocol from " + + "listeners and listener.security.protocol.map (a listener the map does not name has the protocol of its " + + "name, and a configuration without listeners counts as Kafka's default, PLAINTEXT://:9092). Unencrypted " + + "means PLAINTEXT or SASL_PLAINTEXT. SASL counts as on when some listener is SASL_PLAINTEXT or SASL_SSL and " + + "none is PLAINTEXT. An SSL listener counts as neither: a cluster whose clients authenticate only with " + + "certificates reports no SASL, and an SSL listener beside a SASL one passes whether or not it asks for a " + + "certificate. SASL Mechanism Strength reads each SASL listener's " + + "listener.name..sasl.enabled.mechanisms and, when Kafka does not report that key, the " + + "broker-wide sasl.enabled.mechanisms. Whether Kafka reports the per-listener key has not been checked on a " + + "live cluster; where it does not, a Strimzi cluster, which leaves the broker-wide key empty, reports no SASL " + + "mechanisms beside a passing SASL Authentication. The keystore, truststore, mTLS, TLS and certificate " + + "checks still read only broker-wide ssl.* settings (ssl.keystore.location, ssl.truststore.location, " + + "ssl.client.auth, ssl.protocol and the like), never listener.name..* settings. On a cluster that " + + "sets TLS per named listener, such as the Strimzi cluster the kafka-cluster chart installs (listener plain " + + "on 9092, SCRAM without TLS; listener tls on 9093, TLS with client certificates), they can report no " + + "keystore and no client authentication while the tls listener has both: read none of them as the truth " + + "about the listeners.", Refs: []string{ - mcpJava + "service/SecurityService.java:99-126,357-393,685-699", - mcpJava + "service/SecurityService.java:1049-1158,1286-1310", - mcpJava + "service/SecurityPentestService.java:127-137,155-166", + mcpAnchoredRef(mcpJava+"service/SecurityKafkaHelpers.java:74-156", + "listener.security.protocol.map", "static boolean saslRequired(", "static String saslMechanisms("), + mcpAnchoredRef(mcpJava+"service/SecurityService.java:99-132,332-348,365-401,693-703", + `"SASL Authentication"`, `"SASL Mechanism Strength"`, `"SSL Keystore Configured"`, `"Listener Protocol Map"`), + mcpAnchoredRef(mcpJava+"service/SecurityService.java:1053-1162,1290-1314", + "public Map tlsInspect()", "public Map certificateCheck()"), + mcpAnchoredRef(mcpJava+"service/SecurityPentestService.java:127-138,156-168", + `"metadata-leak"`, `"unencrypted"`), "charts/kafka-cluster/values.yaml:90-105", "charts/kafka-cluster/README.md:370-379", }, @@ -983,7 +996,8 @@ var mcpCaveatsSecurity = mcpJoinCaveats([]mcpCaveat{ "has but the fresh audit lacks is not listed. Without a saved baseline there is no drift to report, and " + "this server cannot save one: the baseline is a POST, which its transport refuses.", Refs: []string{ - mcpJava + "service/SecurityService.java:908-975,1010-1017", + mcpAnchoredRef(mcpJava+"service/SecurityService.java:912-979,1014-1021", + "public Map securityDrift()", "private int statusRank(String status)"), mcpJava + "api/SecurityResource.java:134-148", "cli/cmd/mcp_guard.go:1110-1145", }, @@ -994,6 +1008,7 @@ var mcpCaveatsSecurity = mcpJoinCaveats([]mcpCaveat{ "CIS ids are the ones the checks carry, SOC2 and PCI-DSS controls are assigned by check category alone " + "(auth, authz and transport for SOC2; transport, auth and limits for PCI-DSS), and the percentage is the " + "share of mapped checks that pass.", - Refs: []string{mcpJava + "service/SecurityService.java:788-849,1030-1047"}, + Refs: []string{mcpAnchoredRef(mcpJava+"service/SecurityService.java:792-853,1034-1051", + "public Map securityCompliance()", "private String mapToSoc2(String category)")}, }, }, mcpCaveatsScenario) diff --git a/eval/mcp/oracle.py b/eval/mcp/oracle.py index 79badb3b..0be8cd99 100644 --- a/eval/mcp/oracle.py +++ b/eval/mcp/oracle.py @@ -353,7 +353,7 @@ def expect_security_drift(drift: Mapping[str, Any]) -> dict: "improved_checks": sorted(d["check"] for d in changes if d.get("change") == "IMPROVED"), "degraded_checks": sorted(d["check"] for d in changes if d.get("change") == "DEGRADED"), # The grade history is an in-memory list of the last 100 explicit - # audits in the pod (SecurityService.java:42,1649-1689). + # audits in the pod (SecurityService.java:42,1653-1693). "grade_history_shows_trend": False, } diff --git a/kates/src/main/java/com/bmscomp/kates/service/SecurityKafkaHelpers.java b/kates/src/main/java/com/bmscomp/kates/service/SecurityKafkaHelpers.java index ba7c19c4..9479ba0a 100644 --- a/kates/src/main/java/com/bmscomp/kates/service/SecurityKafkaHelpers.java +++ b/kates/src/main/java/com/bmscomp/kates/service/SecurityKafkaHelpers.java @@ -3,9 +3,14 @@ import java.util.ArrayList; import java.util.Collections; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; +import java.util.Locale; import java.util.Map; +import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; +import java.util.stream.Collectors; import jakarta.enterprise.context.ApplicationScoped; import org.apache.kafka.clients.admin.AdminClient; @@ -65,4 +70,88 @@ public String formatBytes(long bytes) { if (bytes >= 1_024) return String.format("%.1fKB", bytes / 1_024.0); return bytes + "B"; } + + /** + * Each listener's security protocol, by listener name, as the broker + * resolves it: the name before :// in listeners, looked up in + * listener.security.protocol.map, or the name itself when the map has no + * entry. The checks read these, not the text of listeners, where Strimzi's + * PLAIN-9092 holds no PLAINTEXT://, and not sasl.enabled.mechanisms, which + * Kafka reports as GSSAPI on a broker with no SASL listener. A config + * without listeners, one that could not be read included, gets Kafka's + * default, PLAINTEXT://:9092. + */ + static Map listenerProtocols(Map config) { + Map protocolMap = new LinkedHashMap<>(); + for (String entry : + config.getOrDefault("listener.security.protocol.map", "").split(",")) { + int colon = entry.indexOf(':'); + if (colon > 0) { + protocolMap.put(normalised(entry.substring(0, colon)), normalised(entry.substring(colon + 1))); + } + } + String listeners = config.getOrDefault("listeners", ""); + Map protocols = new LinkedHashMap<>(); + for (String listener : (listeners.isBlank() ? "PLAINTEXT://:9092" : listeners).split(",")) { + int end = listener.indexOf("://"); + if (end > 0) { + String name = normalised(listener.substring(0, end)); + protocols.put(name, protocolMap.getOrDefault(name, name)); + } + } + return protocols; + } + + /** Kafka upper-cases listener names and protocols before it compares them. */ + private static String normalised(String name) { + return name.trim().toUpperCase(Locale.ROOT); + } + + /** SASL_PLAINTEXT or SASL_SSL: the listener's clients authenticate. */ + static boolean sasl(String protocol) { + return protocol.startsWith("SASL_"); + } + + /** PLAINTEXT or SASL_PLAINTEXT: the listener's traffic is not encrypted. */ + static boolean unencrypted(String protocol) { + return protocol.equals("PLAINTEXT") || protocol.equals("SASL_PLAINTEXT"); + } + + /** + * Every client has to authenticate: some listener is SASL_* and none is + * PLAINTEXT. An SSL listener counts as neither, with or without client + * certificates. + */ + static boolean saslRequired(Map protocols) { + return protocols.values().stream().anyMatch(SecurityKafkaHelpers::sasl) + && !protocols.containsValue("PLAINTEXT"); + } + + /** The listeners whose protocol matches, as "PLAIN-9092 (SASL_PLAINTEXT), ...", or "" when none does. */ + static String listeners(Map protocols, Predicate protocol) { + return protocols.entrySet().stream() + .filter(e -> protocol.test(e.getValue())) + .map(e -> e.getKey() + " (" + e.getValue() + ")") + .collect(Collectors.joining(", ")); + } + + /** + * The mechanisms the SASL listeners enable, joined by commas, or "" when + * no listener is SASL_*: each one's + * listener.name..sasl.enabled.mechanisms, as Strimzi sets them, + * or the broker-wide sasl.enabled.mechanisms when that key is not reported. + */ + static String saslMechanisms(Map config, Map protocols) { + String brokerWide = config.getOrDefault("sasl.enabled.mechanisms", ""); + Set mechanisms = new LinkedHashSet<>(); + protocols.forEach((name, protocol) -> { + if (sasl(protocol)) { + String key = "listener.name." + name.toLowerCase(Locale.ROOT) + ".sasl.enabled.mechanisms"; + for (String mechanism : config.getOrDefault(key, brokerWide).split(",")) { + if (!mechanism.isBlank()) mechanisms.add(mechanism.trim()); + } + } + }); + return String.join(",", mechanisms); + } } diff --git a/kates/src/main/java/com/bmscomp/kates/service/SecurityPentestService.java b/kates/src/main/java/com/bmscomp/kates/service/SecurityPentestService.java index d6b8a21c..193c61b6 100644 --- a/kates/src/main/java/com/bmscomp/kates/service/SecurityPentestService.java +++ b/kates/src/main/java/com/bmscomp/kates/service/SecurityPentestService.java @@ -125,14 +125,15 @@ private Map pentestLargeMessage(Map config) { } private Map pentestMetadataLeak(Map config) { - String sasl = config.getOrDefault("sasl.enabled.mechanisms", ""); - boolean vulnerable = sasl.isEmpty(); + // The audit's SASL Authentication rule, not sasl.enabled.mechanisms: Kafka + // reports GSSAPI there on a broker with no SASL listener + boolean vulnerable = !SecurityKafkaHelpers.saslRequired(SecurityKafkaHelpers.listenerProtocols(config)); return pentestResult( "Unauthenticated Metadata", "metadata-leak", vulnerable, vulnerable - ? "No SASL — metadata request succeeds without credentials" + ? "No SASL listener, or a PLAINTEXT one beside it — metadata request succeeds without credentials" : "SASL required — unauthenticated metadata requests rejected", "HIGH"); } @@ -153,14 +154,15 @@ private Map pentestConnectionLimits(Map config) } private Map pentestUnencryptedAccess(Map config) { - String listeners = config.getOrDefault("listeners", ""); - boolean vulnerable = listeners.contains("PLAINTEXT://"); + String unencrypted = SecurityKafkaHelpers.listeners( + SecurityKafkaHelpers.listenerProtocols(config), SecurityKafkaHelpers::unencrypted); + boolean vulnerable = !unencrypted.isEmpty(); return pentestResult( "Unencrypted Access", "unencrypted", vulnerable, vulnerable - ? "PLAINTEXT listener found — traffic can be sniffed" + ? "Unencrypted listeners: " + unencrypted + " — traffic can be sniffed" : "All listeners use encrypted transport", "HIGH"); } diff --git a/kates/src/main/java/com/bmscomp/kates/service/SecurityService.java b/kates/src/main/java/com/bmscomp/kates/service/SecurityService.java index 06cb6d0f..47c96b7b 100644 --- a/kates/src/main/java/com/bmscomp/kates/service/SecurityService.java +++ b/kates/src/main/java/com/bmscomp/kates/service/SecurityService.java @@ -96,31 +96,37 @@ Map runAudit() { Map brokerConfig = helpers.fetchBrokerConfig(client, brokerId); List acls = helpers.fetchAcls(client); - // 1. SASL/Authentication check - String saslEnabled = brokerConfig.getOrDefault("sasl.enabled.mechanisms", ""); - boolean hasSasl = !saslEnabled.isEmpty(); + // 1. SASL/Authentication check, by listener protocol: Kafka reports + // sasl.enabled.mechanisms=GSSAPI on a broker with no SASL listener + Map protocols = SecurityKafkaHelpers.listenerProtocols(brokerConfig); + String saslListeners = SecurityKafkaHelpers.listeners(protocols, SecurityKafkaHelpers::sasl); + boolean hasSasl = SecurityKafkaHelpers.saslRequired(protocols); checks.add(check( "SASL Authentication", "auth", hasSasl ? "PASS" : "FAIL", hasSasl - ? "SASL mechanisms: " + saslEnabled - : "No SASL mechanisms configured — any client can connect", + ? "SASL listeners: " + saslListeners + : saslListeners.isEmpty() + ? "No SASL listener — any client can connect" + : "PLAINTEXT listeners: " + + SecurityKafkaHelpers.listeners(protocols, "PLAINTEXT"::equals) + + " — any client can connect there", "HIGH", "CIS-4.1", "Set listener.security.protocol.map and sasl.enabled.mechanisms in broker config")); - // 2. Plaintext listener detection - String listeners = brokerConfig.getOrDefault("listeners", ""); - String advertisedListeners = brokerConfig.getOrDefault("advertised.listeners", ""); - boolean hasPlaintext = listeners.contains("PLAINTEXT://") || advertisedListeners.contains("PLAINTEXT://"); + // 2. Plaintext listener detection, by protocol: a listener with a + // name of its own, such as Strimzi's PLAIN-9092, holds no PLAINTEXT:// + String unencryptedListeners = SecurityKafkaHelpers.listeners(protocols, SecurityKafkaHelpers::unencrypted); + boolean hasPlaintext = !unencryptedListeners.isEmpty(); checks.add(check( "No Plaintext Listeners", "transport", hasPlaintext ? "WARN" : "PASS", hasPlaintext - ? "Plaintext listeners detected — credentials sent unencrypted" - : "All listeners use TLS or SASL", + ? "Unencrypted listeners: " + unencryptedListeners + " — traffic sent unencrypted" + : "All listeners use TLS", "HIGH", "CIS-4.2", "Change listener protocol to SASL_SSL in kafka.listeners configuration")); @@ -323,18 +329,20 @@ Map runAudit() { "CIS-5.6", "Minimize super.users — use ACLs for fine-grained access instead")); - // 17. SASL mechanism strength - String saslMechanisms = brokerConfig.getOrDefault("sasl.enabled.mechanisms", ""); + // 17. SASL mechanism strength, of the mechanisms the SASL listeners enable + String saslMechanisms = SecurityKafkaHelpers.saslMechanisms(brokerConfig, protocols); boolean weakSasl = saslMechanisms.contains("PLAIN") && !saslMechanisms.contains("SCRAM"); checks.add(check( "SASL Mechanism Strength", "auth", saslMechanisms.isEmpty() ? "WARN" : weakSasl ? "WARN" : "PASS", - saslMechanisms.isEmpty() - ? "No SASL mechanisms configured" - : weakSasl - ? "PLAIN mechanism only — passwords sent in cleartext" - : "Mechanisms: " + saslMechanisms, + saslListeners.isEmpty() + ? "No SASL listener" + : saslMechanisms.isEmpty() + ? "No SASL mechanisms configured" + : weakSasl + ? "PLAIN mechanism only — passwords sent in cleartext" + : "Mechanisms: " + saslMechanisms, "HIGH", "CIS-4.10", "Use SCRAM-SHA-512 instead of PLAIN for password-based auth")); @@ -682,18 +690,14 @@ Map runAudit() { "CIS-6.6", "Set num.io.threads >= 8 for production workloads")); - // 41. Listener security protocol map - String protocolMap = brokerConfig.getOrDefault("listener.security.protocol.map", ""); - boolean hasPlaintextListener = protocolMap.contains("PLAINTEXT") && !protocolMap.contains("SASL_PLAINTEXT"); + // 41. Listener security protocol map, for the listeners in use: Kafka's + // default map lists all four protocols, whatever the listeners use + String listenersInUse = SecurityKafkaHelpers.listeners(protocols, protocol -> true); checks.add(check( "Listener Protocol Map", "network", - protocolMap.isEmpty() || !hasPlaintextListener ? "PASS" : "WARN", - protocolMap.isEmpty() - ? "Using default listener protocols" - : hasPlaintextListener - ? "PLAINTEXT listeners in protocol map — unencrypted traffic allowed" - : "Protocol map: " + protocolMap, + hasPlaintext ? "WARN" : "PASS", + "Listeners: " + listenersInUse + (hasPlaintext ? " — unencrypted traffic allowed" : ""), "HIGH", "CIS-4.8", "Map all listeners to SASL_SSL or SSL in listener.security.protocol.map")); diff --git a/kates/src/test/java/com/bmscomp/kates/it/SecurityApiIT.java b/kates/src/test/java/com/bmscomp/kates/it/SecurityApiIT.java index 1d849fbc..d211e237 100644 --- a/kates/src/test/java/com/bmscomp/kates/it/SecurityApiIT.java +++ b/kates/src/test/java/com/bmscomp/kates/it/SecurityApiIT.java @@ -78,6 +78,15 @@ void auditGradesEveryCheckAndTheSummaryAddsUp() { // A PLAINTEXT container with no ACLs cannot be a clean bill of health; // if this ever passes, the audit has stopped auditing. assertTrue(failures + warnings > 0, "an unsecured broker must not grade clean"); + + // The container's listeners are all PLAINTEXT, while Kafka reports + // sasl.enabled.mechanisms=GSSAPI: the checks must read the listeners. + assertEquals("FAIL", fieldOf(checks, "name", "SASL Authentication", "status")); + assertEquals("WARN", fieldOf(checks, "name", "No Plaintext Listeners", "status")); + // An unread config also lands here as the default PLAINTEXT://:9092; + // only the container's real listeners and protocol map name BROKER. + assertTrue( + ((String) fieldOf(checks, "name", "No Plaintext Listeners", "detail")).contains("BROKER (PLAINTEXT)")); } @Test @@ -128,6 +137,8 @@ void pentestRunsEveryProbeAndNamesUnknownOnes() { tests.size(), all.getInt("summary.protected") + all.getInt("summary.vulnerable"), "every probe is either protected or vulnerable"); + assertEquals("VULNERABLE", fieldOf(tests, "id", "metadata-leak", "result")); + assertEquals("VULNERABLE", fieldOf(tests, "id", "unencrypted", "result")); given().when() .get("/api/security/pentest?test=auto-create") @@ -319,4 +330,13 @@ void scoreTrendAccumulatesTheAuditsRunInThisProcess() { private static void assertNotBlank(Object value, String what) { assertTrue(value instanceof String s && !s.isBlank(), what + " must be present, got " + value); } + + /** The {@code field} of the entry whose {@code key} is {@code value}, or null when none is. */ + private static Object fieldOf(List> entries, String key, String value, String field) { + return entries.stream() + .filter(e -> value.equals(e.get(key))) + .findFirst() + .map(e -> e.get(field)) + .orElse(null); + } } diff --git a/kates/src/test/java/com/bmscomp/kates/service/SecurityListenerProtocolsTest.java b/kates/src/test/java/com/bmscomp/kates/service/SecurityListenerProtocolsTest.java new file mode 100644 index 00000000..895a5aa0 --- /dev/null +++ b/kates/src/test/java/com/bmscomp/kates/service/SecurityListenerProtocolsTest.java @@ -0,0 +1,184 @@ +package com.bmscomp.kates.service; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyCollection; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.DescribeTopicsResult; +import org.apache.kafka.clients.admin.ListTopicsResult; +import org.apache.kafka.common.KafkaFuture; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +/** + * The audit checks and pentest probes that turn on what the listeners are. + * They read sasl.enabled.mechanisms, which Kafka reports as GSSAPI on a broker + * with no SASL listener, so a PLAINTEXT broker passed SASL Authentication, and + * looked for the text PLAINTEXT:// in listeners, which Strimzi's PLAIN-9092 + * (SCRAM without TLS) never holds, so it passed No Plaintext Listeners. They + * now take each listener's protocol from listener.security.protocol.map. + */ +class SecurityListenerProtocolsTest { + + private static final String DEFAULT_PROTOCOL_MAP = + "PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL"; + + private Map config = Map.of(); + private SecurityService service; + private SecurityPentestService pentest; + + @BeforeEach + void setUp() { + AdminClient client = mock(AdminClient.class); + ListTopicsResult topics = mock(ListTopicsResult.class); + when(topics.names()).thenReturn(KafkaFuture.completedFuture(Set.of())); + when(client.listTopics()).thenReturn(topics); + DescribeTopicsResult described = mock(DescribeTopicsResult.class); + when(described.allTopicNames()).thenReturn(KafkaFuture.completedFuture(Map.of())); + when(client.describeTopics(anyCollection())).thenReturn(described); + KafkaAdminService admin = mock(KafkaAdminService.class); + when(admin.getClient()).thenReturn(client); + + ClusterHealthService health = mock(ClusterHealthService.class); + when(health.describeCluster()).thenReturn(Map.of("brokerCount", 1, "brokers", List.of(Map.of("id", 0)))); + + // A real helpers bean answering with the test's broker config: a mock + // would also stub formatBytes, which the checks print with. + SecurityKafkaHelpers helpers = new SecurityKafkaHelpers() { + @Override + public Map fetchBrokerConfig(AdminClient c, int brokerId) { + return config; + } + }; + service = new SecurityService(admin, health); + service.helpers = helpers; + pentest = new SecurityPentestService(admin, health, helpers); + } + + /** The kafka-cluster chart's Strimzi broker: plain is SCRAM without TLS, tls and replication are TLS. */ + @ParameterizedTest + @ValueSource(strings = {"", "GSSAPI"}) + void strimziChartListeners(String brokerWideMechanisms) { + config = Map.of( + "listeners", + "REPLICATION-9091://0.0.0.0:9091,PLAIN-9092://0.0.0.0:9092,TLS-9093://0.0.0.0:9093", + "listener.security.protocol.map", + "CONTROLPLANE-9090:SSL,REPLICATION-9091:SSL,PLAIN-9092:SASL_PLAINTEXT,TLS-9093:SSL", + "sasl.enabled.mechanisms", + brokerWideMechanisms, + "listener.name.plain-9092.sasl.enabled.mechanisms", + "SCRAM-SHA-512"); + + Map> checks = audit(); + assertStatus(checks, "SASL Authentication", "PASS"); + assertStatus(checks, "No Plaintext Listeners", "WARN"); + assertTrue(detail(checks, "No Plaintext Listeners").contains("PLAIN-9092 (SASL_PLAINTEXT)")); + assertFalse(detail(checks, "No Plaintext Listeners").contains("TLS-9093"), "an SSL listener is encrypted"); + assertStatus(checks, "Listener Protocol Map", "WARN"); + assertStatus(checks, "SASL Mechanism Strength", "PASS"); + assertEquals("Mechanisms: SCRAM-SHA-512", detail(checks, "SASL Mechanism Strength")); + + Map probes = pentest(); + assertEquals("PROTECTED", probes.get("metadata-leak")); + assertEquals("VULNERABLE", probes.get("unencrypted")); + } + + /** A broker that sets only listeners: Kafka reports its default map and GSSAPI. */ + @Test + void vanillaPlaintextDefaultGssapi() { + config = Map.of( + "listeners", "PLAINTEXT://:9092", + "listener.security.protocol.map", DEFAULT_PROTOCOL_MAP, + "sasl.enabled.mechanisms", "GSSAPI"); + + Map> checks = audit(); + assertStatus(checks, "SASL Authentication", "FAIL"); + assertStatus(checks, "SASL Mechanism Strength", "WARN"); + assertEquals("No SASL listener", detail(checks, "SASL Mechanism Strength")); + assertStatus(checks, "No Plaintext Listeners", "WARN"); + assertStatus(checks, "Listener Protocol Map", "WARN"); + + Map probes = pentest(); + assertEquals("VULNERABLE", probes.get("metadata-leak")); + assertEquals("VULNERABLE", probes.get("unencrypted")); + } + + @Test + void saslSslOnly() { + config = Map.of( + "listeners", "SASL_SSL://:9094", + "listener.security.protocol.map", DEFAULT_PROTOCOL_MAP, + "sasl.enabled.mechanisms", "SCRAM-SHA-512"); + + Map> checks = audit(); + assertStatus(checks, "SASL Authentication", "PASS"); + assertStatus(checks, "No Plaintext Listeners", "PASS"); + assertStatus(checks, "Listener Protocol Map", "PASS"); + assertStatus(checks, "SASL Mechanism Strength", "PASS"); + + Map probes = pentest(); + assertEquals("PROTECTED", probes.get("metadata-leak")); + assertEquals("PROTECTED", probes.get("unencrypted")); + } + + /** A PLAINTEXT listener beside a SASL one still lets any client in. */ + @Test + void plaintextBesideSasl() { + config = Map.of( + "listeners", "PLAINTEXT://:9092,SASL_SSL://:9094", + "listener.security.protocol.map", DEFAULT_PROTOCOL_MAP, + "sasl.enabled.mechanisms", "SCRAM-SHA-512"); + + Map> checks = audit(); + assertStatus(checks, "SASL Authentication", "FAIL"); + assertTrue(detail(checks, "SASL Authentication").contains("PLAINTEXT (PLAINTEXT)")); + assertEquals("VULNERABLE", pentest().get("metadata-leak")); + } + + /** The audit's checks by name. runAudit grades an exception F, so a report with an error fails here. */ + private Map> audit() { + Map report = service.runAudit(); + assertFalse(report.containsKey("error"), () -> "the audit failed: " + report.get("error")); + Map> byName = new LinkedHashMap<>(); + for (Object check : (List) report.get("checks")) { + @SuppressWarnings("unchecked") + Map c = (Map) check; + byName.put((String) c.get("name"), c); + } + return byName; + } + + /** Each pentest probe's result by id. */ + private Map pentest() { + Map report = pentest.pentest("all"); + assertFalse(report.containsKey("error"), () -> "the pentest failed: " + report.get("error")); + Map byId = new LinkedHashMap<>(); + for (Object test : (List) report.get("tests")) { + Map t = (Map) test; + byId.put((String) t.get("id"), (String) t.get("result")); + } + return byId; + } + + private static void assertStatus(Map> checks, String name, String status) { + assertEquals( + status, + checks.get(name).get("status"), + name + ": " + checks.get(name).get("detail")); + } + + private static String detail(Map> checks, String name) { + return (String) checks.get(name).get("detail"); + } +} diff --git a/skills/kates-cli/SKILL.md b/skills/kates-cli/SKILL.md index 2fc2df6d..30fcfdd7 100644 --- a/skills/kates-cli/SKILL.md +++ b/skills/kates-cli/SKILL.md @@ -69,7 +69,7 @@ Say what the answer is not: a lab posture check, not audit evidence. - `security pentest` attacks nothing: its checks read that configuration and the ACL list, and an ACL list it could not read passes as PROTECTED. - `security cve` compares against a fixed list of seven Kafka CVEs, the newest from 2024, and never learns the Kafka version, so it reports every CVE PATCHED. - The TLS and certificate checks read broker-wide `ssl.*` settings, never per-listener ones, and open no certificate: expiry and issuer are never read. A missing setting counts as the secure default the check assumes (`ssl.protocol` TLSv1.3), and Keystore Type and Truststore Type always pass. -- The plaintext checks look for the text `PLAINTEXT://`, so a listener with a name of its own (such as `plain`) is never flagged, whatever its protocol, and Listener Protocol Map passes whenever the map holds SASL_PLAINTEXT, which is unencrypted. On a cluster that sets security per listener, as the kafka-cluster chart does (`plain` on 9092 with SCRAM and no TLS, `tls` on 9093 with TLS), the audit can report no SASL, no keystore and no plaintext listener at once: read none of them as the truth about the listeners. +- The SASL, plaintext and Listener Protocol Map checks take each listener's protocol from `listeners` and `listener.security.protocol.map`, so they flag a SASL_PLAINTEXT listener with a name of its own, such as the kafka-cluster chart's `PLAIN-9092` (SCRAM, no TLS), as unencrypted. SASL Mechanism Strength reads the per-listener `sasl.enabled.mechanisms` first; if Kafka does not report it, that cluster shows no SASL mechanisms beside a passing SASL Authentication. The keystore, truststore and mTLS checks still read only broker-wide `ssl.*` settings, so on that chart's cluster they report no keystore and no client authentication while the `tls` listener on 9093 has both. - Drift compares by check name. A check the baseline lacks counts as IMPROVED, whatever its status, and one the baseline has but the fresh audit lacks is not listed. Without a saved baseline there is no drift; do not save one (`security baseline --save`) unless asked, because it replaces the reference. - The compliance mapping relabels the audit's checks by category; it is not an assessment against the frameworks.