Support one-way SPIFFE compatibility with legacy identities

zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md

@@ -1368,6 +1368,50 @@ and [SASL authentication for ZooKeeper](https://cwiki.apache.org/confluence/disp
     authenticated client with that principal will be able to bypass
     ACL checking and have full privileges to all znodes.
 
+* *ssl.x509.legacySuperUserCompatibilityEnabled* :
+    (Java system property: **zookeeper.ssl.x509.legacySuperUserCompatibilityEnabled**)
+    **Default: false.** When enabled, both **X509AuthenticationProvider** and
+    **X509ZNodeGroupAclProvider** can match an authenticated SPIFFE v1/wl identity,
+    or a v1/v2 **application/\<mp\>/\<app\>[/\<tag\>]** identity, against an
+    existing formatted legacy service-principal superuser ID by application name. Exact
+    configured client-ID matches take precedence and do not require this option.
+    Supported legacy forms include **servicePrincipal(kafka**,
+    **servicePrincipal(kafka)**, and **urn:li:servicePrincipal(kafka;region1;instance1)**.
+    The existing superuser settings remain **zookeeper.X509AuthenticationProvider.superUser**
+    and **zookeeper.X509ZNodeGroupAclProvider.superUserId**, respectively.
+    The original certificate-derived identity is preserved; the **super** AuthInfo
+    marker uses the matched configured ID so explicit superusers remain distinct
+    from cross-domain components during ACL preparation.
+    This option does not enable reverse legacy-to-SPIFFE superuser aliases or
+    compatibility for user, group, airflow, arbitrary v2, or Subject DN identities.
+    **Warning:** legacy application names do not distinguish products or tags;
+    enable this option only when all eligible trusted identities with that app
+    name should have full superuser privileges. It does not change certificate
+    trust validation or existing cross-domain grants. Treat it as a startup
+    setting and restart servers or reconnect clients when changing it; it is not
+    an immediate revocation mechanism for already authenticated connections.
+
+    URI-domain and direct ACL compatibility does not require this option.
+    That compatibility is one-way: eligible SPIFFE application identities may
+    match formatted legacy service principals, but
+    legacy clients do not acquire reverse aliases.
+    Original client IDs, exact matches and existing mapped-domain grants remain
+    unchanged. Bare names such as **kafka** are not compatibility targets for
+    URI-domain mappings, direct ACLs or superuser selection. For example,
+    **application/example-mp/kafka** does not gain an alias to **kafka**.
+    This does not reject existing exact matches: a v1/wl or legacy SAN identity
+    extracted as **kafka** still matches that exact ID. A client explicitly
+    mapped into domain **kafka** can still match the **x509:kafka** domain ACL.
+
+    Direct ACL matching uses authenticated identity context attached to the request,
+    including writes forwarded by followers or observers. Quorum requests carry this
+    context in a reserved transport-only **zookeeper-internal-x509** entry, removed
+    before authorization; it is not a client authentication scheme or a stored ACL.
+    Both the receiving server and the leader must support this context for forwarded
+    compatibility matches. Missing context does not enable an alias, so do not rely
+    on this compatibility during a mixed-version rollout. Exact IDs and existing
+    domain/superuser AuthInfo continue to use their ordinary authorization rules.
+
 * *zookeeper.superUser* :
     (Java system property: **zookeeper.superUser**)
     Similar to **zookeeper.X509AuthenticationProvider.superUser**

zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java

@@ -268,11 +268,11 @@ public void processRequest(Request request) {
                         Record rec;
                         switch (readOp.getType()) {
                         case OpCode.getChildren:
-                            rec = handleGetChildrenRequest(readOp.toRequestRecord(), cnxn, request.authInfo);
+                            rec = handleGetChildrenRequest(readOp.toRequestRecord(), request);
                             subResult = new GetChildrenResult(((GetChildrenResponse) rec).getChildren());
                             break;
                         case OpCode.getData:
-                            rec = handleGetDataRequest(readOp.toRequestRecord(), cnxn, request.authInfo);
+                            rec = handleGetDataRequest(readOp.toRequestRecord(), request);
                             GetDataResponse gdr = (GetDataResponse) rec;
                             subResult = new GetDataResult(gdr.getData(), gdr.getStat());
                             break;
@@ -369,7 +369,7 @@ public void processRequest(Request request) {
                 GetDataRequest getDataRequest = new GetDataRequest();
                 ByteBufferInputStream.byteBuffer2Record(request.request, getDataRequest);
                 path = getDataRequest.getPath();
-                rsp = handleGetDataRequest(getDataRequest, cnxn, request.authInfo);
+                rsp = handleGetDataRequest(getDataRequest, request);
                 requestPathMetricsCollector.registerRequest(request.type, path);
                 break;
             }
@@ -426,9 +426,9 @@ public void processRequest(Request request) {
                     throw new KeeperException.NoNodeException();
                 }
                 zks.checkACL(
-                    request.cnxn,
+                    request,
                     zks.getZKDatabase().aclForNode(n),
-                    ZooDefs.Perms.READ | ZooDefs.Perms.ADMIN, request.authInfo, path,
+                    ZooDefs.Perms.READ | ZooDefs.Perms.ADMIN, path,
                     null);
 
                 Stat stat = new Stat();
@@ -437,10 +437,9 @@ public void processRequest(Request request) {
 
                 try {
                     zks.checkACL(
-                        request.cnxn,
+                        request,
                         zks.getZKDatabase().aclForNode(n),
                         ZooDefs.Perms.ADMIN,
-                        request.authInfo,
                         path,
                         null);
                     rsp = new GetACLResponse(acl, stat);
@@ -464,7 +463,7 @@ public void processRequest(Request request) {
                 GetChildrenRequest getChildrenRequest = new GetChildrenRequest();
                 ByteBufferInputStream.byteBuffer2Record(request.request, getChildrenRequest);
                 path = getChildrenRequest.getPath();
-                rsp = handleGetChildrenRequest(getChildrenRequest, cnxn, request.authInfo);
+                rsp = handleGetChildrenRequest(getChildrenRequest, request);
                 requestPathMetricsCollector.registerRequest(request.type, path);
                 break;
             }
@@ -478,10 +477,9 @@ public void processRequest(Request request) {
                     throw new KeeperException.NoNodeException();
                 }
                 zks.checkACL(
-                    request.cnxn,
+                    request,
                     zks.getZKDatabase().aclForNode(n),
                     ZooDefs.Perms.READ,
-                    request.authInfo,
                     path,
                     null);
                 int number = zks.getZKDatabase().getAllChildrenNumber(path);
@@ -499,10 +497,10 @@ public void processRequest(Request request) {
                     throw new KeeperException.NoNodeException();
                 }
                 zks.checkACL(
-                    request.cnxn,
+                    request,
                     zks.getZKDatabase().aclForNode(n),
                     ZooDefs.Perms.READ,
-                    request.authInfo, path,
+                    path,
                     null);
                 List<String> children = zks.getZKDatabase()
                                            .getChildren(path, stat, getChildren2Request.getWatch() ? cnxn : null);
@@ -569,10 +567,10 @@ public void processRequest(Request request) {
                     throw new KeeperException.NoNodeException();
                 }
                 zks.checkACL(
-                        request.cnxn,
+                        request,
                         zks.getZKDatabase().aclForNode(n),
                         ZooDefs.Perms.READ,
-                        request.authInfo, path,
+                        path,
                         null);
                 final int maxReturned = getChildrenPaginatedRequest.getMaxReturned();
                 final PaginationNextPage nextPage = new PaginationNextPage();
@@ -674,29 +672,29 @@ public void processRequest(Request request) {
         }
     }
 
-    private Record handleGetChildrenRequest(Record request, ServerCnxn cnxn, List<Id> authInfo) throws KeeperException, IOException {
-        GetChildrenRequest getChildrenRequest = (GetChildrenRequest) request;
+    private Record handleGetChildrenRequest(Record record, Request request) throws KeeperException, IOException {
+        GetChildrenRequest getChildrenRequest = (GetChildrenRequest) record;
         String path = getChildrenRequest.getPath();
         DataNode n = zks.getZKDatabase().getNode(path);
         if (n == null) {
             throw new KeeperException.NoNodeException();
         }
-        zks.checkACL(cnxn, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ, authInfo, path, null);
+        zks.checkACL(request, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ, path, null);
         List<String> children = zks.getZKDatabase()
-                                   .getChildren(path, null, getChildrenRequest.getWatch() ? cnxn : null);
+                                   .getChildren(path, null, getChildrenRequest.getWatch() ? request.cnxn : null);
         return new GetChildrenResponse(children);
     }
 
-    private Record handleGetDataRequest(Record request, ServerCnxn cnxn, List<Id> authInfo) throws KeeperException, IOException {
-        GetDataRequest getDataRequest = (GetDataRequest) request;
+    private Record handleGetDataRequest(Record record, Request request) throws KeeperException, IOException {
+        GetDataRequest getDataRequest = (GetDataRequest) record;
         String path = getDataRequest.getPath();
         DataNode n = zks.getZKDatabase().getNode(path);
         if (n == null) {
             throw new KeeperException.NoNodeException();
         }
-        zks.checkACL(cnxn, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ, authInfo, path, null);
+        zks.checkACL(request, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ, path, null);
         Stat stat = new Stat();
-        byte[] b = zks.getZKDatabase().getData(path, stat, getDataRequest.getWatch() ? cnxn : null);
+        byte[] b = zks.getZKDatabase().getData(path, stat, getDataRequest.getWatch() ? request.cnxn : null);
         return new GetDataResponse(b, stat);
     }
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java

@@ -368,7 +368,7 @@ protected void pRequest2Txn(int type, long zxid, Request request, Record record,
             String path = deleteRequest.getPath();
             String parentPath = getParentPathAndValidate(path);
             ChangeRecord parentRecord = getRecordForPath(parentPath);
-            zks.checkACL(request.cnxn, parentRecord.acl, ZooDefs.Perms.DELETE, request.authInfo, path, null);
+            zks.checkACL(request, parentRecord.acl, ZooDefs.Perms.DELETE, path, null);
             ChangeRecord nodeRecord = getRecordForPath(path);
             checkAndIncVersion(nodeRecord.stat.getVersion(), deleteRequest.getVersion(), path);
             if (nodeRecord.childCount > 0) {
@@ -396,7 +396,7 @@ protected void pRequest2Txn(int type, long zxid, Request request, Record record,
             path = setDataRequest.getPath();
             validatePath(path, request.sessionId);
             nodeRecord = getRecordForPath(path);
-            zks.checkACL(request.cnxn, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo, path, null);
+            zks.checkACL(request, nodeRecord.acl, ZooDefs.Perms.WRITE, path, null);
             int newVersion = checkAndIncVersion(nodeRecord.stat.getVersion(), setDataRequest.getVersion(), path);
             request.setTxn(new SetDataTxn(path, setDataRequest.getData(), newVersion));
             nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
@@ -536,7 +536,7 @@ protected void pRequest2Txn(int type, long zxid, Request request, Record record,
             }
 
             nodeRecord = getRecordForPath(ZooDefs.CONFIG_NODE);
-            zks.checkACL(request.cnxn, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo, null, null);
+            zks.checkACL(request, nodeRecord.acl, ZooDefs.Perms.WRITE, null, null);
             SetDataTxn setDataTxn = new SetDataTxn(ZooDefs.CONFIG_NODE, request.qv.toString().getBytes(), -1);
             request.setTxn(setDataTxn);
             nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
@@ -562,7 +562,7 @@ protected void pRequest2Txn(int type, long zxid, Request request, Record record,
             validatePath(path, request.sessionId);
             List<ACL> listACL = fixupACL(path, request.authInfo, setAclRequest.getAcl());
             nodeRecord = getRecordForPath(path);
-            zks.checkACL(request.cnxn, nodeRecord.acl, ZooDefs.Perms.ADMIN, request.authInfo, path, listACL);
+            zks.checkACL(request, nodeRecord.acl, ZooDefs.Perms.ADMIN, path, listACL);
             newVersion = checkAndIncVersion(nodeRecord.stat.getAversion(), setAclRequest.getVersion(), path);
             request.setTxn(new SetACLTxn(path, listACL, newVersion));
             nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
@@ -633,7 +633,7 @@ protected void pRequest2Txn(int type, long zxid, Request request, Record record,
             path = checkVersionRequest.getPath();
             validatePath(path, request.sessionId);
             nodeRecord = getRecordForPath(path);
-            zks.checkACL(request.cnxn, nodeRecord.acl, ZooDefs.Perms.READ, request.authInfo, path, null);
+            zks.checkACL(request, nodeRecord.acl, ZooDefs.Perms.READ, path, null);
             request.setTxn(new CheckVersionTxn(
                 path,
                 checkAndIncVersion(nodeRecord.stat.getVersion(), checkVersionRequest.getVersion(), path)));
@@ -682,7 +682,7 @@ private void pRequest2TxnCreate(int type, Request request, Record record, boolea
         List<ACL> listACL = fixupACL(path, request.authInfo, acl);
         ChangeRecord parentRecord = getRecordForPath(parentPath);
 
-        zks.checkACL(request.cnxn, parentRecord.acl, ZooDefs.Perms.CREATE, request.authInfo, path, listACL);
+        zks.checkACL(request, parentRecord.acl, ZooDefs.Perms.CREATE, path, listACL);
         int parentCVersion = parentRecord.stat.getCversion();
         if (createMode.isSequential()) {
             path = path + String.format(Locale.ENGLISH, "%010d", parentCVersion);

zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java

@@ -27,6 +27,7 @@
 import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.metrics.Summary;
 import org.apache.zookeeper.metrics.SummarySet;
+import org.apache.zookeeper.server.auth.X509AuthenticationUtil.ClientIdentity;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 import org.apache.zookeeper.server.util.AuthUtil;
 import org.apache.zookeeper.txn.TxnDigest;
@@ -50,12 +51,18 @@ public class Request {
     private static volatile boolean staleLatencyCheck = Boolean.parseBoolean(System.getProperty("zookeeper.request_stale_latency_check", "false"));
 
     public Request(ServerCnxn cnxn, long sessionId, int xid, int type, ByteBuffer bb, List<Id> authInfo) {
+        this(cnxn, sessionId, xid, type, bb, authInfo, cnxn == null ? null : cnxn.getX509ClientIdentity());
+    }
+
+    public Request(ServerCnxn cnxn, long sessionId, int xid, int type, ByteBuffer bb, List<Id> authInfo,
+                   ClientIdentity x509ClientIdentity) {
         this.cnxn = cnxn;
         this.sessionId = sessionId;
         this.cxid = xid;
         this.type = type;
         this.request = bb;
         this.authInfo = authInfo;
+        this.x509ClientIdentity = x509ClientIdentity;
     }
 
     public Request(long sessionId, int xid, int type, TxnHeader hdr, Record txn, long zxid) {
@@ -68,6 +75,7 @@ public Request(long sessionId, int xid, int type, TxnHeader hdr, Record txn, lon
         this.request = null;
         this.cnxn = null;
         this.authInfo = null;
+        this.x509ClientIdentity = null;
     }
 
     public final long sessionId;
@@ -88,6 +96,12 @@ public Request(long sessionId, int xid, int type, TxnHeader hdr, Record txn, lon
 
     public final List<Id> authInfo;
 
+    private final ClientIdentity x509ClientIdentity;
+
+    public ClientIdentity getX509ClientIdentity() {
+        return x509ClientIdentity;
+    }
+
     public final long createTime = Time.currentElapsedTime();
 
     public long prepQueueStartTime = -1;

zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxn.java

@@ -46,6 +46,7 @@
 import org.apache.zookeeper.metrics.Counter;
 import org.apache.zookeeper.proto.ReplyHeader;
 import org.apache.zookeeper.proto.RequestHeader;
+import org.apache.zookeeper.server.auth.X509AuthenticationUtil.ClientIdentity;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -62,6 +63,9 @@ public abstract class ServerCnxn implements Stats, Watcher {
 
     private Set<Id> authInfo = Collections.newSetFromMap(new ConcurrentHashMap<Id, Boolean>());
 
+    // Retain authenticated type information without reparsing certificates during ACL checks.
+    private volatile ClientIdentity x509ClientIdentity;
+
     private static final byte[] fourBytes = new byte[4];
 
     /**
@@ -285,6 +289,14 @@ public boolean removeAuthInfo(Id id) {
         return authInfo.remove(id);
     }
 
+    public ClientIdentity getX509ClientIdentity() {
+        return x509ClientIdentity;
+    }
+
+    public void setX509ClientIdentity(ClientIdentity identity) {
+        x509ClientIdentity = identity;
+    }
+
     abstract void sendBuffer(ByteBuffer... buffers);
 
     abstract void enableRecv();

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java

@@ -1985,6 +1985,18 @@ public void dumpMonitorValues(BiConsumer<String, Object> response) {
      * @param setAcls : for set ACL operations, the list of ACLs being set. Otherwise null.
      */
     public void checkACL(ServerCnxn cnxn, List<ACL> acl, int perm, List<Id> ids, String path, List<ACL> setAcls) throws KeeperException.NoAuthException {
+        checkACLWithContext(new ServerAuthenticationProvider.ServerObjs(this, cnxn), acl, perm, ids, path, setAcls);
+    }
+
+    public void checkACL(Request request, List<ACL> acl, int perm, String path, List<ACL> setAcls)
+        throws KeeperException.NoAuthException {
+        checkACLWithContext(
+            new ServerAuthenticationProvider.ServerObjs(this, request.cnxn, request.getX509ClientIdentity()),
+            acl, perm, request.authInfo, path, setAcls);
+    }
+
+    private void checkACLWithContext(ServerAuthenticationProvider.ServerObjs serverObjs, List<ACL> acl, int perm,
+                                    List<Id> ids, String path, List<ACL> setAcls) throws KeeperException.NoAuthException {
         if (skipACL) {
             return;
         }
@@ -2012,7 +2024,7 @@ public void checkACL(ServerCnxn cnxn, List<ACL> acl, int perm, List<Id> ids, Str
                     for (Id authId : ids) {
                         if (authId.getScheme().equals(id.getScheme())
                             && ap.matches(
-                                new ServerAuthenticationProvider.ServerObjs(this, cnxn),
+                                serverObjs,
                                 new ServerAuthenticationProvider.MatchValues(path, authId.getId(), id.getId(), perm, setAcls))) {
                             return;
                         }
@@ -2142,7 +2154,7 @@ public boolean authWriteRequest(Request request) {
         try {
             pathToCheck = effectiveACLPath(request);
             if (pathToCheck != null) {
-                checkACL(request.cnxn, zkDb.getACL(pathToCheck, null), effectiveACLPerms(request), request.authInfo, pathToCheck, null);
+                checkACL(request, zkDb.getACL(pathToCheck, null), effectiveACLPerms(request), pathToCheck, null);
             }
         } catch (KeeperException.NoAuthException e) {
             LOG.debug("Request failed ACL check", e);