diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java index 4eb5d93dbdc4e..63ca4cdea7011 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java @@ -141,6 +141,26 @@ CompletableFuture canLookupAsync(TopicName topicName, String role, CompletableFuture allowFunctionOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData); + /** + * Allow all source operations with in this namespace + * @param namespaceName The namespace that the sources operations can be executed in + * @param role The role to check + * @param authenticationData authentication data related to the role + * @return a boolean to determine whether authorized or not + */ + CompletableFuture allowSourceOpsAsync(NamespaceName namespaceName, String role, + AuthenticationDataSource authenticationData); + + /** + * Allow all sink operations with in this namespace + * @param namespaceName The namespace that the sink operations can be executed in + * @param role The role to check + * @param authenticationData authentication data related to the role + * @return a boolean to determine whether authorized or not + */ + CompletableFuture allowSinkOpsAsync(NamespaceName namespaceName, String role, + AuthenticationDataSource authenticationData); + /** * * Grant authorization-action permission on a namespace to the given client diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java index 10b35ef7ffedc..e964faa966d23 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java @@ -331,6 +331,16 @@ public CompletableFuture allowFunctionOpsAsync(NamespaceName namespaceN return provider.allowFunctionOpsAsync(namespaceName, role, authenticationData); } + public CompletableFuture allowSourceOpsAsync(NamespaceName namespaceName, String role, + AuthenticationDataSource authenticationData) { + return provider.allowSourceOpsAsync(namespaceName, role, authenticationData); + } + + public CompletableFuture allowSinkOpsAsync(NamespaceName namespaceName, String role, + AuthenticationDataSource authenticationData) { + return provider.allowSinkOpsAsync(namespaceName, role, authenticationData); + } + /** * Grant authorization-action permission on a tenant to the given client * diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java index 1aa79bf965ae4..a3943113520ff 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java @@ -221,6 +221,22 @@ public CompletableFuture canLookupAsync(TopicName topicName, String rol @Override public CompletableFuture allowFunctionOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) { + return allowFunctionSourceSinkOpsAsync(namespaceName, role, authenticationData, AuthAction.functions); + } + + @Override + public CompletableFuture allowSourceOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) { + return allowFunctionSourceSinkOpsAsync(namespaceName, role, authenticationData, AuthAction.sources); + } + + @Override + public CompletableFuture allowSinkOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) { + return allowFunctionSourceSinkOpsAsync(namespaceName, role, authenticationData, AuthAction.sinks); + } + + private CompletableFuture allowFunctionSourceSinkOpsAsync(NamespaceName namespaceName, String role, + AuthenticationDataSource authenticationData, + AuthAction authAction) { CompletableFuture permissionFuture = new CompletableFuture<>(); try { configCache.policiesCache().getAsync(POLICY_ROOT + namespaceName.toString()).thenAccept(policies -> { @@ -231,7 +247,7 @@ public CompletableFuture allowFunctionOpsAsync(NamespaceName namespaceN } else { Map> namespaceRoles = policies.get().auth_policies.namespace_auth; Set namespaceActions = namespaceRoles.get(role); - if (namespaceActions != null && namespaceActions.contains(AuthAction.functions)) { + if (namespaceActions != null && namespaceActions.contains(authAction)) { // The role has namespace level permission permissionFuture.complete(true); return; @@ -239,7 +255,7 @@ public CompletableFuture allowFunctionOpsAsync(NamespaceName namespaceN // Using wildcard if (conf.isAuthorizationAllowWildcardsMatching()) { - if (checkWildcardPermission(role, AuthAction.functions, namespaceRoles)) { + if (checkWildcardPermission(role, authAction, namespaceRoles)) { // The role has namespace level permission by wildcard match permissionFuture.complete(true); return; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java index ddeed965eb523..180142c05a036 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java @@ -469,6 +469,16 @@ public CompletableFuture allowFunctionOpsAsync(NamespaceName namespaceN return null; } + @Override + public CompletableFuture allowSourceOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) { + return null; + } + + @Override + public CompletableFuture allowSinkOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) { + return null; + } + @Override public CompletableFuture grantPermissionAsync(NamespaceName namespace, Set actions, String role, String authenticationData) { diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/AuthAction.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/AuthAction.java index 6f70e964e0944..646ca03f95283 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/AuthAction.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/AuthAction.java @@ -30,4 +30,10 @@ public enum AuthAction { /** Permissions for functions ops. **/ functions, + + /** Permissions for sources ops. **/ + sources, + + /** Permissions for sinks ops. **/ + sinks, } diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java index 2e31f7387784c..11c7b20cc2396 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java @@ -1518,8 +1518,18 @@ && worker().getWorkerConfig().getSuperUserRoles() != null public boolean allowFunctionOps(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) { try { - return worker().getAuthorizationService().allowFunctionOpsAsync( - namespaceName, role, authenticationData).get(worker().getWorkerConfig().getZooKeeperOperationTimeoutSeconds(), SECONDS); + switch (componentType) { + case SINK: + return worker().getAuthorizationService().allowSinkOpsAsync( + namespaceName, role, authenticationData).get(worker().getWorkerConfig().getZooKeeperOperationTimeoutSeconds(), SECONDS); + case SOURCE: + return worker().getAuthorizationService().allowSourceOpsAsync( + namespaceName, role, authenticationData).get(worker().getWorkerConfig().getZooKeeperOperationTimeoutSeconds(), SECONDS); + case FUNCTION: + default: + return worker().getAuthorizationService().allowFunctionOpsAsync( + namespaceName, role, authenticationData).get(worker().getWorkerConfig().getZooKeeperOperationTimeoutSeconds(), SECONDS); + } } catch (InterruptedException e) { log.warn("Time-out {} sec while checking function authorization on {} ", worker().getWorkerConfig().getZooKeeperOperationTimeoutSeconds(), namespaceName); throw new RestException(Status.INTERNAL_SERVER_ERROR, e.getMessage()); diff --git a/site2/website/release-notes.md b/site2/website/release-notes.md index 2476a674facba..5bb0d49126125 100644 --- a/site2/website/release-notes.md +++ b/site2/website/release-notes.md @@ -1,6 +1,12 @@ ## Apache Pulsar Release Notes +### 2.7.0 — Not Yet Released + +##### Upgrade notes + +* [IO] If Function Authorization is enabled, users have to be given the source/sink entitlement to run them. See https://github.com/apache/pulsar/pull/7466 + ### 2.6.0 — 2020-06-17 #### Features