Support mTLS connection-level capabilities and RPC access filtering in C++

Adds the following:
 * Named capabilities and capability sets that represent (respectively)
   a single Vespa access API (such as Document API, search API etc)
   or a concrete subset of individual capabilities that make up a
   particular Vespa service (such as a content node).
 * A new `capabilities` array field to the mTLS authorization policies
   that allows for constraining what requests sent over a particular
   connection are allowed to actually do. Capabilities are referenced
   by name and may include any combination of capability sets and
   individual capabilities. If multiple capabilities/sets are configured,
   the resulting set of capabilities is the union set of all of them.
 * An FRT RPC-level access filter that can be set up as part of RPC
   method definitions. If set, filters are invoked prior to RPC methods.
 * A new `PERMISSION_DENIED` error code to FRT RPC that is invoked if
   an access filter denies a request.

This also GCs the unused `AssumedRoles` concept which is now deprecated
in favor of capabilities.

Note: this is **not yet** a public or stable API, and capability
names/semantics may change at any time.

5 files changed, 293 insertions, 69 deletions

diff --git a/vespalib/src/tests/net/tls/capabilities/CMakeLists.txt b/vespalib/src/tests/net/tls/capabilities/CMakeLists.txt
new file mode 100644
index 00000000000..4e366674d36
--- /dev/null
+++ b/vespalib/src/tests/net/tls/capabilities/CMakeLists.txt
@@ -0,0 +1,10 @@
+# Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(vespalib_net_tls_capabilities_test_app TEST
+    SOURCES
+    capabilities_test.cpp
+    DEPENDS
+    vespalib
+)
+vespa_add_test(NAME vespalib_net_tls_capabilities_test_app
+               COMMAND vespalib_net_tls_capabilities_test_app)
+
diff --git a/vespalib/src/tests/net/tls/capabilities/capabilities_test.cpp b/vespalib/src/tests/net/tls/capabilities/capabilities_test.cpp
new file mode 100644
index 00000000000..5f74bdceff4
--- /dev/null
+++ b/vespalib/src/tests/net/tls/capabilities/capabilities_test.cpp
@@ -0,0 +1,152 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/vespalib/net/tls/capability_set.h>
+#include <vespa/vespalib/testkit/test_kit.h>
+
+using namespace vespalib;
+using namespace vespalib::net::tls;
+using namespace std::string_view_literals;
+
+TEST("Capability bit positions are stable across calls") {
+    auto cap1 = Capability::content_storage_api();
+    auto cap2 = Capability::content_storage_api();
+    EXPECT_EQUAL(cap1.id_bit_pos(), cap2.id_bit_pos());
+}
+
+TEST("Capability instances are equality comparable") {
+    auto cap1 = Capability::content_document_api();
+    auto cap2 = Capability::content_document_api();
+    auto cap3 = Capability::content_storage_api();
+    EXPECT_EQUAL(cap1, cap2);
+    EXPECT_EQUAL(cap2, cap1);
+    EXPECT_NOT_EQUAL(cap1, cap3);
+}
+
+TEST("Can get underlying name of all Capability instances") {
+    EXPECT_EQUAL(Capability::content_storage_api().name(),  "vespa.content.storage_api"sv);
+    EXPECT_EQUAL(Capability::content_document_api().name(), "vespa.content.document_api"sv);
+    EXPECT_EQUAL(Capability::content_search_api().name(),   "vespa.content.search_api"sv);
+    EXPECT_EQUAL(Capability::slobrok_api().name(),          "vespa.slobrok.api"sv);
+    EXPECT_EQUAL(Capability::content_status_pages().name(), "vespa.content.status_pages"sv);
+    EXPECT_EQUAL(Capability::content_metrics_api().name(),  "vespa.content.metrics_api"sv);
+    EXPECT_EQUAL(Capability::content_cluster_controller_internal_state_api().name(),
+                 "vespa.content.cluster_controller.internal_state_api"sv);
+}
+
+TEST("Capability instances can be stringified") {
+    EXPECT_EQUAL(Capability::content_storage_api().to_string(), "Capability(vespa.content.storage_api)");
+}
+
+TEST("All known capabilities can be looked up by name") {
+    EXPECT_TRUE(Capability::find_capability("vespa.content.storage_api").has_value());
+    EXPECT_TRUE(Capability::find_capability("vespa.content.document_api").has_value());
+    EXPECT_TRUE(Capability::find_capability("vespa.content.search_api").has_value());
+    EXPECT_TRUE(Capability::find_capability("vespa.content.cluster_controller.internal_state_api").has_value());
+    EXPECT_TRUE(Capability::find_capability("vespa.slobrok.api").has_value());
+    EXPECT_TRUE(Capability::find_capability("vespa.content.status_pages").has_value());
+    EXPECT_TRUE(Capability::find_capability("vespa.content.metrics_api").has_value());
+}
+
+TEST("Unknown capability name returns nullopt") {
+    EXPECT_FALSE(Capability::find_capability("vespa.content.stale_cat_memes").has_value());
+}
+
+TEST("CapabilitySet instances can be stringified") {
+    EXPECT_EQUAL(CapabilitySet::content_node().to_string(),
+                 "CapabilitySet({vespa.content.storage_api, vespa.content.document_api, vespa.slobrok.api})");
+}
+
+TEST("All known capability sets can be looked up by name") {
+    EXPECT_TRUE(CapabilitySet::find_capability_set("vespa.content_node").has_value());
+    EXPECT_TRUE(CapabilitySet::find_capability_set("vespa.container_node").has_value());
+    EXPECT_TRUE(CapabilitySet::find_capability_set("vespa.telemetry").has_value());
+    EXPECT_TRUE(CapabilitySet::find_capability_set("vespa.cluster_controller_node").has_value());
+    EXPECT_TRUE(CapabilitySet::find_capability_set("vespa.config_server").has_value());
+}
+
+TEST("Unknown capability set name returns nullopt") {
+    EXPECT_FALSE(CapabilitySet::find_capability_set("vespa.unicorn_launcher").has_value());
+}
+
+TEST("Resolving a capability set adds all its underlying capabilities") {
+    CapabilitySet caps;
+    EXPECT_TRUE(caps.resolve_and_add("vespa.content_node"));
+    // Slightly suboptimal; this test will fail if the default set of capabilities for vespa.content_node changes.
+    EXPECT_EQUAL(caps.count(), 3u);
+    EXPECT_FALSE(caps.empty());
+    EXPECT_TRUE(caps.contains(Capability::content_storage_api()));
+    EXPECT_TRUE(caps.contains(Capability::content_document_api()));
+    EXPECT_TRUE(caps.contains(Capability::slobrok_api()));
+    EXPECT_FALSE(caps.contains(Capability::content_search_api()));
+}
+
+TEST("Resolving a single capability adds it to the underlying capabilities") {
+    CapabilitySet caps;
+    EXPECT_TRUE(caps.resolve_and_add("vespa.slobrok.api"));
+    EXPECT_EQUAL(caps.count(), 1u);
+    EXPECT_FALSE(caps.empty());
+    EXPECT_TRUE(caps.contains(Capability::slobrok_api()));
+    EXPECT_FALSE(caps.contains(Capability::content_storage_api()));
+}
+
+TEST("Resolving an unknown capability set returns false and does not add anything") {
+    CapabilitySet caps;
+    EXPECT_FALSE(caps.resolve_and_add("vespa.distributors_evil_twin_with_an_evil_goatee"));
+    EXPECT_EQUAL(caps.count(), 0u);
+    EXPECT_TRUE(caps.empty());
+}
+
+TEST("Default-constructed CapabilitySet has no capabilities") {
+    CapabilitySet caps;
+    EXPECT_EQUAL(caps.count(), 0u);
+    EXPECT_TRUE(caps.empty());
+    EXPECT_FALSE(caps.contains(Capability::content_storage_api()));
+}
+
+TEST("CapabilitySet can be created with all capabilities") {
+    auto caps = CapabilitySet::make_with_all_capabilities();
+    EXPECT_EQUAL(caps.count(), max_capability_bit_count());
+    EXPECT_TRUE(caps.contains(Capability::content_storage_api()));
+    EXPECT_TRUE(caps.contains(Capability::content_metrics_api()));
+    // ... we just assume the rest are present as well.
+}
+
+TEST("CapabilitySet::contains_all() requires an intersection of capabilities") {
+    auto cap1 = Capability::content_document_api();
+    auto cap2 = Capability::content_search_api();
+    auto cap3 = Capability::content_storage_api();
+
+    const auto all_caps = CapabilitySet::make_with_all_capabilities();
+    auto set_123 = CapabilitySet::of({cap1, cap2, cap3});
+    auto set_13  = CapabilitySet::of({cap1, cap3});
+    auto set_2   = CapabilitySet::of({cap2});
+    auto set_23  = CapabilitySet::of({cap2, cap3});
+    auto empty   = CapabilitySet::make_empty();
+
+    // Sets contain themselves
+    EXPECT_TRUE(all_caps.contains_all(all_caps));
+    EXPECT_TRUE(set_13.contains_all(set_13));
+    EXPECT_TRUE(set_2.contains_all(set_2));
+    EXPECT_TRUE(empty.contains_all(empty));
+
+    // Supersets contain subsets
+    EXPECT_TRUE(all_caps.contains_all(set_123));
+    EXPECT_TRUE(all_caps.contains_all(set_13));
+    EXPECT_TRUE(set_123.contains_all(set_13));
+    EXPECT_TRUE(set_2.contains_all(empty));
+
+    // Subsets do not contain supersets
+    EXPECT_FALSE(set_123.contains_all(all_caps));
+    EXPECT_FALSE(set_13.contains_all(set_123));
+    EXPECT_FALSE(empty.contains_all(set_2));
+
+    // Partially overlapping sets are not contained in each other
+    EXPECT_FALSE(set_13.contains_all(set_23));
+    EXPECT_FALSE(set_23.contains_all(set_13));
+
+    // Fully disjoint sets are not contained in each other
+    EXPECT_FALSE(set_2.contains_all(set_13));
+    EXPECT_FALSE(set_13.contains_all(set_2));
+}
+
+TEST_MAIN() { TEST_RUN_ALL(); }
diff --git a/vespalib/src/tests/net/tls/openssl_impl/openssl_impl_test.cpp b/vespalib/src/tests/net/tls/openssl_impl/openssl_impl_test.cpp
index 1de10939bea..3d19c335c19 100644
--- a/vespalib/src/tests/net/tls/openssl_impl/openssl_impl_test.cpp
+++ b/vespalib/src/tests/net/tls/openssl_impl/openssl_impl_test.cpp
@@ -542,7 +542,7 @@ struct PrintingCertificateCallback : CertificateVerificationCallback {
         for (auto& dns : peer_creds.dns_sans) {
             fprintf(stderr, "Got a DNS SAN entry: %s\n", dns.c_str());
         }
-        return VerificationResult::make_authorized_for_all_roles();
+        return VerificationResult::make_authorized_with_all_capabilities();
     }
 };
 
@@ -551,7 +551,7 @@ struct MockCertificateCallback : CertificateVerificationCallback {
     mutable PeerCredentials creds; // only used in single thread testing context
     VerificationResult verify(const PeerCredentials& peer_creds) const override {
         creds = peer_creds;
-        return VerificationResult::make_authorized_for_all_roles();
+        return VerificationResult::make_authorized_with_all_capabilities();
     }
 };
 
@@ -712,6 +712,29 @@ TEST_F("Server allows client with certificate that DOES match peer policy", Cert
     EXPECT_TRUE(f.handshake());
 }
 
+TEST_F("Authz policy-derived peer capabilities are propagated to CryptoCodec", CertFixture) {
+    auto server_ck = f.create_ca_issued_peer_cert({}, {{"DNS:hello.world.example.com"}});
+    auto authorized = authorized_peers({policy_with({required_san_dns("stale.memes.example.com")},
+                                                    CapabilitySet::of({Capability::content_search_api(),
+                                                                       Capability::content_status_pages()})),
+                                        policy_with({required_san_dns("fresh.memes.example.com")},
+                                                    CapabilitySet::make_with_all_capabilities())});
+    f.reset_server_with_cert_opts(server_ck, std::move(authorized));
+    auto client_ck = f.create_ca_issued_peer_cert({}, {{"DNS:stale.memes.example.com"}});
+    f.reset_client_with_cert_opts(client_ck, AuthorizedPeers::allow_all_authenticated());
+
+    ASSERT_TRUE(f.handshake());
+
+    // Note: "inversion" of client <-> server is because the capabilities are that of the _peer_.
+    auto client_caps = f.server->granted_capabilities();
+    auto server_caps = f.client->granted_capabilities();
+    // Server (from client's PoV) implicitly has all capabilities since client doesn't specify any policies
+    EXPECT_EQUAL(server_caps, CapabilitySet::make_with_all_capabilities());
+    // Client (from server's PoV) only has capabilities for the rule matching its DNS SAN entry
+    EXPECT_EQUAL(client_caps, CapabilitySet::of({Capability::content_search_api(),
+                                                 Capability::content_status_pages()}));
+}
+
 void reset_peers_with_server_authz_mode(CertFixture& f, AuthorizationMode authz_mode) {
     auto ck = f.create_ca_issued_peer_cert({"hello.world.example.com"}, {});
 
diff --git a/vespalib/src/tests/net/tls/policy_checking_certificate_verifier/policy_checking_certificate_verifier_test.cpp b/vespalib/src/tests/net/tls/policy_checking_certificate_verifier/policy_checking_certificate_verifier_test.cpp
index fa2bc1a2eaf..c456d7e2a5c 100644
--- a/vespalib/src/tests/net/tls/policy_checking_certificate_verifier/policy_checking_certificate_verifier_test.cpp
+++ b/vespalib/src/tests/net/tls/policy_checking_certificate_verifier/policy_checking_certificate_verifier_test.cpp
@@ -127,9 +127,9 @@ bool verify(AuthorizedPeers authorized_peers, const PeerCredentials& peer_creds)
     return verifier->verify(peer_creds).success();
 }
 
-AssumedRoles verify_roles(AuthorizedPeers authorized_peers, const PeerCredentials& peer_creds) {
+CapabilitySet verify_capabilities(AuthorizedPeers authorized_peers, const PeerCredentials& peer_creds) {
     auto verifier = create_verify_callback_from(std::move(authorized_peers));
-    return verifier->verify(peer_creds).steal_assumed_roles();
+    return verifier->verify(peer_creds).granted_capabilities();
 }
 
 TEST("Default-constructed AuthorizedPeers does not allow all authenticated peers") {
@@ -142,14 +142,16 @@ TEST("Specially constructed set of policies allows all authenticated peers") {
     EXPECT_TRUE(verify(allow_all, creds_with_dns_sans({{"anything.goes"}})));
 }
 
-TEST("specially constructed set of policies returns wildcard role set") {
+TEST("specially constructed set of policies returns full capability set") {
     auto allow_all = AuthorizedPeers::allow_all_authenticated();
-    EXPECT_EQUAL(verify_roles(allow_all, creds_with_dns_sans({{"anything.goes"}})), AssumedRoles::make_wildcard_role());
+    EXPECT_EQUAL(verify_capabilities(allow_all, creds_with_dns_sans({{"anything.goes"}})),
+                 CapabilitySet::make_with_all_capabilities());
 }
 
-TEST("policy without explicit role set implicitly returns wildcard role set") {
+TEST("policy without explicit capability set implicitly returns full capability set") {
     auto authorized = authorized_peers({policy_with({required_san_dns("yolo.swag")})});
-    EXPECT_EQUAL(verify_roles(authorized, creds_with_dns_sans({{"yolo.swag"}})), AssumedRoles::make_wildcard_role());
+    EXPECT_EQUAL(verify_capabilities(authorized, creds_with_dns_sans({{"yolo.swag"}})),
+                 CapabilitySet::make_with_all_capabilities());
 }
 
 TEST("Non-empty policies do not allow all authenticated peers") {
@@ -246,11 +248,11 @@ struct MultiPolicyMatchFixture {
 };
 
 MultiPolicyMatchFixture::MultiPolicyMatchFixture()
-    : authorized(authorized_peers({policy_with({required_san_dns("hello.world")},   assumed_roles({"r1"})),
-                                   policy_with({required_san_dns("foo.bar")},       assumed_roles({"r2"})),
-                                   policy_with({required_san_dns("zoid.berg")},     assumed_roles({"r2", "r3"})),
-                                   policy_with({required_san_dns("secret.sauce")},  AssumedRoles::make_wildcard_role()),
-                                   policy_with({required_san_uri("zoid://be.rg/")}, assumed_roles({"r4"}))}))
+    : authorized(authorized_peers({policy_with({required_san_dns("hello.world")},   CapabilitySet::of({cap_1()})),
+                                   policy_with({required_san_dns("foo.bar")},       CapabilitySet::of({cap_2()})),
+                                   policy_with({required_san_dns("zoid.berg")},     CapabilitySet::of({cap_2(), cap_3()})),
+                                   policy_with({required_san_dns("secret.sauce")},  CapabilitySet::make_with_all_capabilities()),
+                                   policy_with({required_san_uri("zoid://be.rg/")}, CapabilitySet::of({cap_4()}))}))
 {}
 
 MultiPolicyMatchFixture::~MultiPolicyMatchFixture() = default;
@@ -262,32 +264,37 @@ TEST_F("peer verifies if it matches at least 1 policy of multiple", MultiPolicyM
     EXPECT_TRUE(verify(f.authorized, creds_with_uri_sans({{"zoid://be.rg/"}})));
 }
 
-TEST_F("role set is returned for single matched policy", MultiPolicyMatchFixture) {
-    EXPECT_EQUAL(verify_roles(f.authorized, creds_with_dns_sans({{"hello.world"}})),   assumed_roles({"r1"}));
-    EXPECT_EQUAL(verify_roles(f.authorized, creds_with_dns_sans({{"foo.bar"}})),       assumed_roles({"r2"}));
-    EXPECT_EQUAL(verify_roles(f.authorized, creds_with_dns_sans({{"zoid.berg"}})),     assumed_roles({"r2", "r3"}));
-    EXPECT_EQUAL(verify_roles(f.authorized, creds_with_dns_sans({{"secret.sauce"}})),  AssumedRoles::make_wildcard_role());
-    EXPECT_EQUAL(verify_roles(f.authorized, creds_with_uri_sans({{"zoid://be.rg/"}})), assumed_roles({"r4"}));
+TEST_F("capability set is returned for single matched policy", MultiPolicyMatchFixture) {
+    EXPECT_EQUAL(verify_capabilities(f.authorized, creds_with_dns_sans({{"hello.world"}})),
+                 CapabilitySet::of({cap_1()}));
+    EXPECT_EQUAL(verify_capabilities(f.authorized, creds_with_dns_sans({{"foo.bar"}})),
+                 CapabilitySet::of({cap_2()}));
+    EXPECT_EQUAL(verify_capabilities(f.authorized, creds_with_dns_sans({{"zoid.berg"}})),
+                 CapabilitySet::of({cap_2(), cap_3()}));
+    EXPECT_EQUAL(verify_capabilities(f.authorized, creds_with_dns_sans({{"secret.sauce"}})),
+                 CapabilitySet::make_with_all_capabilities());
+    EXPECT_EQUAL(verify_capabilities(f.authorized, creds_with_uri_sans({{"zoid://be.rg/"}})),
+                 CapabilitySet::of({cap_4()}));
 }
 
 TEST_F("peer verifies if it matches multiple policies", MultiPolicyMatchFixture) {
     EXPECT_TRUE(verify(f.authorized, creds_with_dns_sans({{"hello.world"}, {"zoid.berg"}})));
 }
 
-TEST_F("union role set is returned if multiple policies match", MultiPolicyMatchFixture) {
-    EXPECT_EQUAL(verify_roles(f.authorized, creds_with_dns_sans({{"hello.world"}, {"foo.bar"}, {"zoid.berg"}})),
-                 assumed_roles({"r1", "r2", "r3"}));
-    // Wildcard role is tracked as a distinct role string
-    EXPECT_EQUAL(verify_roles(f.authorized, creds_with_dns_sans({{"hello.world"}, {"foo.bar"}, {"secret.sauce"}})),
-                 assumed_roles({"r1", "r2", "*"}));
+TEST_F("union capability set is returned if multiple policies match", MultiPolicyMatchFixture) {
+    EXPECT_EQUAL(verify_capabilities(f.authorized, creds_with_dns_sans({{"hello.world"}, {"foo.bar"}, {"zoid.berg"}})),
+                 CapabilitySet::of({cap_1(), cap_2(), cap_3()}));
+    EXPECT_EQUAL(verify_capabilities(f.authorized, creds_with_dns_sans({{"hello.world"}, {"foo.bar"}, {"secret.sauce"}})),
+                 CapabilitySet::make_with_all_capabilities());
 }
 
 TEST_F("peer must match at least 1 of multiple policies", MultiPolicyMatchFixture) {
     EXPECT_FALSE(verify(f.authorized, creds_with_dns_sans({{"does.not.exist"}})));
 }
 
-TEST_F("empty role set is returned if no policies match", MultiPolicyMatchFixture) {
-    EXPECT_EQUAL(verify_roles(f.authorized, creds_with_dns_sans({{"does.not.exist"}})), AssumedRoles::make_empty());
+TEST_F("empty capability set is returned if no policies match", MultiPolicyMatchFixture) {
+    EXPECT_EQUAL(verify_capabilities(f.authorized, creds_with_dns_sans({{"does.not.exist"}})),
+                 CapabilitySet::make_empty());
 }
 
 TEST("CN requirement without glob pattern is matched as exact string") {
@@ -308,62 +315,32 @@ TEST("CN requirement can include glob wildcards") {
     EXPECT_FALSE(verify(authorized, creds_with_cn("world")));
 }
 
-TEST("AssumedRoles by default contains no roles") {
-    AssumedRoles roles;
-    EXPECT_TRUE(roles.empty());
-    EXPECT_FALSE(roles.can_assume_role("foo"));
-    auto empty = AssumedRoles::make_empty();
-    EXPECT_EQUAL(roles, empty);
-}
-
-TEST("AssumedRoles can be constructed with an explicit set of roles") {
-    auto roles = AssumedRoles::make_for_roles({"foo", "bar"});
-    EXPECT_TRUE(roles.can_assume_role("foo"));
-    EXPECT_TRUE(roles.can_assume_role("bar"));
-    EXPECT_FALSE(roles.can_assume_role("baz"));
-}
-
-TEST("AssumedRoles wildcard role can assume any role") {
-    auto roles = AssumedRoles::make_wildcard_role();
-    EXPECT_TRUE(roles.can_assume_role("foo"));
-    EXPECT_TRUE(roles.can_assume_role("bar"));
-}
-
-TEST("AssumedRolesBuilder builds union set of added roles") {
-    AssumedRolesBuilder builder;
-    builder.add_union(AssumedRoles::make_for_roles({"hello", "world"}));
-    builder.add_union(AssumedRoles::make_for_roles({"hello", "moon"}));
-    builder.add_union(AssumedRoles::make_for_roles({"goodbye", "moon"}));
-    auto roles = builder.build_with_move();
-    EXPECT_EQUAL(roles, AssumedRoles::make_for_roles({"hello", "goodbye", "moon", "world"}));
-}
-
 TEST("VerificationResult is not authorized by default") {
     VerificationResult result;
     EXPECT_FALSE(result.success());
-    EXPECT_TRUE(result.assumed_roles().empty());
+    EXPECT_TRUE(result.granted_capabilities().empty());
 }
 
 TEST("VerificationResult can be explicitly created as not authorized") {
     auto result = VerificationResult::make_not_authorized();
     EXPECT_FALSE(result.success());
-    EXPECT_TRUE(result.assumed_roles().empty());
+    EXPECT_TRUE(result.granted_capabilities().empty());
 }
 
-TEST("VerificationResult can be pre-authorized for all roles") {
-    auto result = VerificationResult::make_authorized_for_all_roles();
+TEST("VerificationResult can be pre-authorized with all capabilities") {
+    auto result = VerificationResult::make_authorized_with_all_capabilities();
     EXPECT_TRUE(result.success());
-    EXPECT_FALSE(result.assumed_roles().empty());
-    EXPECT_TRUE(result.assumed_roles().can_assume_role("foo"));
+    EXPECT_FALSE(result.granted_capabilities().empty());
+    EXPECT_EQUAL(result.granted_capabilities(), CapabilitySet::make_with_all_capabilities());
 }
 
-TEST("VerificationResult can be pre-authorized for an explicit set of roles") {
-    auto result = VerificationResult::make_authorized_for_roles(AssumedRoles::make_for_roles({"elden", "ring"}));
+TEST("VerificationResult can be pre-authorized for an explicit set of capabilities") {
+    auto result = VerificationResult::make_authorized_with_capabilities(CapabilitySet::of({cap_2(), cap_3()}));
     EXPECT_TRUE(result.success());
-    EXPECT_FALSE(result.assumed_roles().empty());
-    EXPECT_TRUE(result.assumed_roles().can_assume_role("elden"));
-    EXPECT_TRUE(result.assumed_roles().can_assume_role("ring"));
-    EXPECT_FALSE(result.assumed_roles().can_assume_role("O you don't have the right"));
+    EXPECT_FALSE(result.granted_capabilities().empty());
+    EXPECT_TRUE(result.granted_capabilities().contains(cap_2()));
+    EXPECT_TRUE(result.granted_capabilities().contains(cap_3()));
+    EXPECT_FALSE(result.granted_capabilities().contains(cap_1()));
 }
 
 // TODO test CN _and_ SAN
diff --git a/vespalib/src/tests/net/tls/transport_options/transport_options_reading_test.cpp b/vespalib/src/tests/net/tls/transport_options/transport_options_reading_test.cpp
index be2c63b03f2..8d49bdbf73d 100644
--- a/vespalib/src/tests/net/tls/transport_options/transport_options_reading_test.cpp
+++ b/vespalib/src/tests/net/tls/transport_options/transport_options_reading_test.cpp
@@ -198,6 +198,68 @@ TEST("unknown fields are ignored at parse-time") {
     EXPECT_TRUE(read_options_from_json_string(json).get() != nullptr); // And no exception thrown.
 }
 
+TEST("policy without explicit capabilities implicitly get all capabilities") {
+    const char* json = R"({
+      "required-credentials":[
+         {"field": "SAN_DNS", "must-match": "hello.world"}
+      ]
+    })";
+    EXPECT_EQUAL(authorized_peers({policy_with({required_san_dns("hello.world")},
+                                               CapabilitySet::make_with_all_capabilities())}),
+                 parse_policies(json).authorized_peers());
+}
+
+TEST("specifying a capability set adds all its underlying capabilities") {
+    const char* json = R"({
+      "required-credentials":[
+         {"field": "SAN_DNS", "must-match": "*.cool-content-clusters.example" }
+      ],
+      "capabilities": ["vespa.content_node"]
+    })";
+    EXPECT_EQUAL(authorized_peers({policy_with({required_san_dns("*.cool-content-clusters.example")},
+                                               CapabilitySet::content_node())}),
+                 parse_policies(json).authorized_peers());
+}
+
+TEST("can specify single leaf capabilities") {
+    const char* json = R"({
+      "required-credentials":[
+         {"field": "SAN_DNS", "must-match": "*.cool-content-clusters.example" }
+      ],
+      "capabilities": ["vespa.content.metrics_api", "vespa.slobrok.api"]
+    })";
+    EXPECT_EQUAL(authorized_peers({policy_with({required_san_dns("*.cool-content-clusters.example")},
+                                               CapabilitySet::of({Capability::content_metrics_api(),
+                                                                  Capability::slobrok_api()}))}),
+                 parse_policies(json).authorized_peers());
+}
+
+TEST("specifying multiple capability sets adds union of underlying capabilities") {
+    const char* json = R"({
+      "required-credentials":[
+         {"field": "SAN_DNS", "must-match": "*.cool-content-clusters.example" }
+      ],
+      "capabilities": ["vespa.content_node", "vespa.container_node"]
+    })";
+    CapabilitySet caps;
+    caps.add_all(CapabilitySet::content_node());
+    caps.add_all(CapabilitySet::container_node());
+    EXPECT_EQUAL(authorized_peers({policy_with({required_san_dns("*.cool-content-clusters.example")}, caps)}),
+                 parse_policies(json).authorized_peers());
+}
+
+TEST("empty capabilities array is not allowed") {
+    const char* json = R"({
+      "required-credentials":[
+         {"field": "SAN_DNS", "must-match": "*.cool-content-clusters.example" }
+      ],
+      "capabilities": []
+    })";
+    EXPECT_EXCEPTION(parse_policies(json), vespalib::IllegalArgumentException,
+                     "\"capabilities\" array must either be not present (implies "
+                     "all capabilities) or contain at least one capability name");
+}
+
 // TODO test parsing of multiple policies
 
 TEST_MAIN() { TEST_RUN_ALL(); }