Skip to content

Commit e2ddc4b

Browse files
mziccardmziccard
committed
Add support for Pub/Sub IAM methods and tests
1 parent ba74dc3 commit e2ddc4b

11 files changed

Lines changed: 1114 additions & 3 deletions

File tree

‎google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/PolicyMarshaller.java‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
/*
2+
* Copyright 2016 Google Inc. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.pubsub;
18+
19+
import com.google.cloud.Policy;
20+
21+
final class PolicyMarshaller extends Policy.DefaultMarshaller {
22+
23+
static final PolicyMarshaller INSTANCE = new PolicyMarshaller();
24+
25+
private PolicyMarshaller() {}
26+
27+
@Override
28+
protected com.google.iam.v1.Policy toPb(Policy policy) {
29+
return super.toPb(policy);
30+
}
31+
32+
@Override
33+
protected Policy fromPb(com.google.iam.v1.Policy policyPb) {
34+
return super.fromPb(policyPb);
35+
}
36+
}

‎google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/PubSub.java‎

Lines changed: 176 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import com.google.cloud.AsyncPage;
2020
import com.google.cloud.GrpcServiceOptions.ExecutorFactory;
2121
import com.google.cloud.Page;
22+
import com.google.cloud.Policy;
2223
import com.google.cloud.Service;
2324

2425
import java.util.Iterator;
@@ -655,7 +656,179 @@ Future<Void> modifyAckDeadlineAsync(String subscription, int deadline, TimeUnit
655656
Future<Void> modifyAckDeadlineAsync(String subscription, int deadline, TimeUnit unit,
656657
Iterable<String> ackIds);
657658

658-
// IAM Policy operations: getPolicy, replacePolicy, testPermissions
659-
// Not sure if ready (docs is not up-to-date)
660-
// Looks like policy is per resource (topic or subscription) but not per service?
659+
/**
660+
* Returns the IAM access control policy for the specified topic. Returns {@code null} if the
661+
* topic was not found.
662+
*
663+
* @throws PubSubException upon failure
664+
*/
665+
Policy getTopicPolicy(String topic);
666+
667+
/**
668+
* Sends a request for getting the IAM access control policy for the specified topic. This method
669+
* returns a {@code Future} object to consume the result. {@link Future#get()} returns the
670+
* requested policy or {@code null} if the topic was not found.
671+
*
672+
* @throws PubSubException upon failure
673+
*/
674+
Future<Policy> getTopicPolicyAsync(String topic);
675+
676+
/**
677+
* Sets the IAM access control policy for the specified topic. Replaces any existing policy. This
678+
* method returns the new policy.
679+
*
680+
* <p>It is recommended that you use the read-modify-write pattern. This pattern entails reading
681+
* the project's current policy, updating it locally, and then sending the modified policy for
682+
* writing. Cloud IAM solves the problem of conflicting processes simultaneously attempting to
683+
* modify a policy by using the {@link Policy#etag etag} property. This property is used to
684+
* verify whether the policy has changed since the last request. When you make a request with an
685+
* etag value, the value in the request is compared with the existing etag value associated with
686+
* the policy. The policy is written only if the etag values match. If the etags don't match, a
687+
* {@code PubSubException} is thrown, denoting that the server aborted update. If an etag is not
688+
* provided, the policy is overwritten blindly.
689+
*
690+
* @throws PubSubException upon failure
691+
*/
692+
Policy replaceTopicPolicy(String topic, Policy newPolicy);
693+
694+
/**
695+
* Sends a request to set the IAM access control policy for the specified topic. Replaces any
696+
* existing policy. This method returns a {@code Future} object to consume the result.
697+
* {@link Future#get()} returns the new policy.
698+
*
699+
* <p>It is recommended that you use the read-modify-write pattern. This pattern entails reading
700+
* the project's current policy, updating it locally, and then sending the modified policy for
701+
* writing. Cloud IAM solves the problem of conflicting processes simultaneously attempting to
702+
* modify a policy by using the {@link Policy#etag etag} property. This property is used to
703+
* verify whether the policy has changed since the last request. When you make a request with an
704+
* etag value, the value in the request is compared with the existing etag value associated with
705+
* the policy. The policy is written only if the etag values match. If the etags don't match,
706+
* {@link Future#get()} will throw a {@link java.util.concurrent.ExecutionException} caused by a
707+
* {@code PubSubException}, denoting that the server aborted update. If an etag is not provided,
708+
* the policy is overwritten blindly.
709+
*
710+
* @throws PubSubException upon failure
711+
*/
712+
Future<Policy> replaceTopicPolicyAsync(String topic, Policy newPolicy);
713+
714+
/**
715+
* Returns the permissions that a caller has on the specified topic.
716+
*
717+
* <p>You typically don't call this method if you're using Google Cloud Platform directly to
718+
* manage permissions. This method is intended for integration with your proprietary software,
719+
* such as a customized graphical user interface. For example, the Cloud Platform Console tests
720+
* IAM permissions internally to determine which UI should be available to the logged-in user.
721+
*
722+
* @return A list of booleans representing whether the caller has the permissions specified (in
723+
* the order of the given permissions)
724+
* @throws PubSubException upon failure
725+
* @see <a href="https://cloud.google.com/pubsub/docs/access_control#permissions">
726+
* Permissions and Roles</a>
727+
*/
728+
List<Boolean> testTopicPermissions(String topic, List<String> permissions);
729+
730+
/**
731+
* Sends a request to get the permissions that a caller has on the specified topic.
732+
*
733+
* <p>You typically don't call this method if you're using Google Cloud Platform directly to
734+
* manage permissions. This method is intended for integration with your proprietary software,
735+
* such as a customized graphical user interface. For example, the Cloud Platform Console tests
736+
* IAM permissions internally to determine which UI should be available to the logged-in user.
737+
*
738+
* @return A {@code Future} object to consume the result. {@link Future#get()} returns a list of
739+
* booleans representing whether the caller has the permissions specified (in the order of the
740+
* given permissions)
741+
* @throws PubSubException upon failure
742+
* @see <a href="https://cloud.google.com/pubsub/docs/access_control#permissions">
743+
* Permissions and Roles</a>
744+
*/
745+
Future<List<Boolean>> testTopicPermissionsAsync(String topic, List<String> permissions);
746+
747+
/**
748+
* Returns the IAM access control policy for the specified subscription. Returns {@code null} if
749+
* the subscription was not found.
750+
*
751+
* @throws PubSubException upon failure
752+
*/
753+
Policy getSubscriptionPolicy(String subscription);
754+
755+
/**
756+
* Sends a request for getting the IAM access control policy for the specified subscription. This
757+
* method returns a {@code Future} object to consume the result. {@link Future#get()} returns the
758+
* requested policy or {@code null} if the subscription was not found.
759+
*
760+
* @throws PubSubException upon failure
761+
*/
762+
Future<Policy> getSubscriptionPolicyAsync(String subscription);
763+
764+
/**
765+
* Sets the IAM access control policy for the specified subscription. Replaces any existing
766+
* policy. This method returns the new policy.
767+
*
768+
* <p>It is recommended that you use the read-modify-write pattern. This pattern entails reading
769+
* the project's current policy, updating it locally, and then sending the modified policy for
770+
* writing. Cloud IAM solves the problem of conflicting processes simultaneously attempting to
771+
* modify a policy by using the {@link Policy#etag etag} property. This property is used to
772+
* verify whether the policy has changed since the last request. When you make a request with an
773+
* etag value, the value in the request is compared with the existing etag value associated with
774+
* the policy. The policy is written only if the etag values match. If the etags don't match, a
775+
* {@code PubSubException} is thrown, denoting that the server aborted update. If an etag is not
776+
* provided, the policy is overwritten blindly.
777+
*
778+
* @throws PubSubException upon failure
779+
*/
780+
Policy replaceSubscriptionPolicy(String subscription, Policy newPolicy);
781+
782+
/**
783+
* Sends a request to set the IAM access control policy for the specified subscription. Replaces
784+
* any existing policy. This method returns a {@code Future} object to consume the result.
785+
* {@link Future#get()} returns the new policy.
786+
*
787+
* <p>It is recommended that you use the read-modify-write pattern. This pattern entails reading
788+
* the project's current policy, updating it locally, and then sending the modified policy for
789+
* writing. Cloud IAM solves the problem of conflicting processes simultaneously attempting to
790+
* modify a policy by using the {@link Policy#etag etag} property. This property is used to
791+
* verify whether the policy has changed since the last request. When you make a request with an
792+
* etag value, the value in the request is compared with the existing etag value associated with
793+
* the policy. The policy is written only if the etag values match. If the etags don't match,
794+
* {@link Future#get()} will throw a {@link java.util.concurrent.ExecutionException} caused by a
795+
* {@code PubSubException}, denoting that the server aborted update. If an etag is not provided,
796+
* the policy is overwritten blindly.
797+
*
798+
* @throws PubSubException upon failure
799+
*/
800+
Future<Policy> replaceSubscriptionPolicyAsync(String subscription, Policy newPolicy);
801+
802+
/**
803+
* Returns the permissions that a caller has on the specified subscription. You typically don't
804+
* call this method if you're using Google Cloud Platform directly to manage permissions. This
805+
* method is intended for integration with your proprietary software, such as a customized
806+
* graphical user interface. For example, the Cloud Platform Console tests IAM permissions
807+
* internally to determine which UI should be available to the logged-in user.
808+
*
809+
* @return A list of booleans representing whether the caller has the permissions specified (in
810+
* the order of the given permissions)
811+
* @throws PubSubException upon failure
812+
* @see <a href="https://cloud.google.com/pubsub/docs/access_control#permissions">
813+
* Permissions and Roles</a>
814+
*/
815+
List<Boolean> testSubscriptionPermissions(String subscription, List<String> permissions);
816+
817+
/**
818+
* Sends a request to get the permissions that a caller has on the specified subscription.
819+
*
820+
* <p>You typically don't call this method if you're using Google Cloud Platform directly to
821+
* manage permissions. This method is intended for integration with your proprietary software,
822+
* such as a customized graphical user interface. For example, the Cloud Platform Console tests
823+
* IAM permissions internally to determine which UI should be available to the logged-in user.
824+
*
825+
* @return A {@code Future} object to consume the result. {@link Future#get()} returns a list of
826+
* booleans representing whether the caller has the permissions specified (in the order of the
827+
* given permissions)
828+
* @throws PubSubException upon failure
829+
* @see <a href="https://cloud.google.com/pubsub/docs/access_control#permissions">
830+
* Permissions and Roles</a>
831+
*/
832+
Future<List<Boolean>> testSubscriptionPermissionsAsync(String subscription,
833+
List<String> permissions);
661834
}

‎google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/PubSubImpl.java‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,13 +20,15 @@
2020
import static com.google.cloud.pubsub.PubSub.ListOption.OptionType.PAGE_TOKEN;
2121
import static com.google.cloud.pubsub.PubSub.PullOption.OptionType.EXECUTOR_FACTORY;
2222
import static com.google.cloud.pubsub.PubSub.PullOption.OptionType.MAX_QUEUED_CALLBACKS;
23+
import static com.google.common.base.MoreObjects.firstNonNull;
2324
import static com.google.common.base.Preconditions.checkArgument;
2425

2526
import com.google.cloud.AsyncPage;
2627
import com.google.cloud.AsyncPageImpl;
2728
import com.google.cloud.BaseService;
2829
import com.google.cloud.Page;
2930
import com.google.cloud.PageImpl;
31+
import com.google.cloud.Policy;
3032
import com.google.cloud.pubsub.spi.PubSubRpc;
3133
import com.google.cloud.pubsub.spi.PubSubRpc.PullFuture;
3234
import com.google.cloud.pubsub.spi.v1.PublisherApi;
@@ -35,13 +37,17 @@
3537
import com.google.common.base.Function;
3638
import com.google.common.base.Throwables;
3739
import com.google.common.collect.ImmutableList;
40+
import com.google.common.collect.ImmutableSet;
3841
import com.google.common.collect.Iterables;
3942
import com.google.common.collect.Iterators;
4043
import com.google.common.collect.Lists;
4144
import com.google.common.collect.Maps;
4245
import com.google.common.util.concurrent.Futures;
4346
import com.google.common.util.concurrent.ListenableFuture;
4447
import com.google.common.util.concurrent.Uninterruptibles;
48+
import com.google.iam.v1.SetIamPolicyRequest;
49+
import com.google.iam.v1.TestIamPermissionsRequest;
50+
import com.google.iam.v1.TestIamPermissionsResponse;
4551
import com.google.protobuf.Empty;
4652
import com.google.pubsub.v1.AcknowledgeRequest;
4753
import com.google.pubsub.v1.DeleteSubscriptionRequest;
@@ -65,6 +71,7 @@
6571
import java.util.Iterator;
6672
import java.util.List;
6773
import java.util.Map;
74+
import java.util.Set;
6875
import java.util.concurrent.ExecutionException;
6976
import java.util.concurrent.Future;
7077
import java.util.concurrent.TimeUnit;
@@ -95,6 +102,13 @@ public String apply(com.google.pubsub.v1.ReceivedMessage message) {
95102
return message.getAckId();
96103
}
97104
};
105+
private static final Function<com.google.iam.v1.Policy, Policy> POLICY_TO_PB_FUNCTION =
106+
new Function<com.google.iam.v1.Policy, Policy>() {
107+
@Override
108+
public Policy apply(com.google.iam.v1.Policy policyPb) {
109+
return policyPb == null ? null : PolicyMarshaller.INSTANCE.fromPb(policyPb);
110+
}
111+
};
98112

99113
PubSubImpl(PubSubOptions options) {
100114
super(options);
@@ -601,6 +615,102 @@ public Future<Void> modifyAckDeadlineAsync(String subscription, int deadline, Ti
601615
return transform(rpc.modify(request), EMPTY_TO_VOID_FUNCTION);
602616
}
603617

618+
@Override
619+
public Policy getTopicPolicy(String topic) {
620+
return get(getTopicPolicyAsync(topic));
621+
}
622+
623+
@Override
624+
public Future<Policy> getTopicPolicyAsync(String topic) {
625+
return transform(rpc.getIamPolicy(PublisherApi.formatTopicName(options().projectId(), topic)),
626+
POLICY_TO_PB_FUNCTION);
627+
}
628+
629+
@Override
630+
public Policy replaceTopicPolicy(String topic, Policy newPolicy) {
631+
return get(replaceTopicPolicyAsync(topic, newPolicy));
632+
}
633+
634+
@Override
635+
public Future<Policy> replaceTopicPolicyAsync(String topic, Policy newPolicy) {
636+
SetIamPolicyRequest request = SetIamPolicyRequest.newBuilder()
637+
.setPolicy(PolicyMarshaller.INSTANCE.toPb(newPolicy))
638+
.setResource(PublisherApi.formatTopicName(options().projectId(), topic))
639+
.build();
640+
return transform(rpc.setIamPolicy(request), POLICY_TO_PB_FUNCTION);
641+
}
642+
643+
@Override
644+
public List<Boolean> testTopicPermissions(String topic, final List<String> permissions) {
645+
return get(testTopicPermissionsAsync(topic, permissions));
646+
}
647+
648+
@Override
649+
public Future<List<Boolean>> testTopicPermissionsAsync(String topic, List<String> permissions) {
650+
TestIamPermissionsRequest request = TestIamPermissionsRequest.newBuilder()
651+
.setResource(PublisherApi.formatTopicName(options().projectId(), topic))
652+
.addAllPermissions(permissions)
653+
.build();
654+
return transform(rpc.testIamPermissions(request), permissionsFromPbFunction(permissions));
655+
}
656+
657+
@Override
658+
public Policy getSubscriptionPolicy(String subscription) {
659+
return get(getSubscriptionPolicyAsync(subscription));
660+
}
661+
662+
@Override
663+
public Future<Policy> getSubscriptionPolicyAsync(String subscription) {
664+
return transform(
665+
rpc.getIamPolicy(SubscriberApi.formatSubscriptionName(options().projectId(), subscription)),
666+
POLICY_TO_PB_FUNCTION);
667+
}
668+
669+
@Override
670+
public Policy replaceSubscriptionPolicy(String subscription, Policy newPolicy) {
671+
return get(replaceSubscriptionPolicyAsync(subscription, newPolicy));
672+
}
673+
674+
@Override
675+
public Future<Policy> replaceSubscriptionPolicyAsync(String subscription, Policy newPolicy) {
676+
SetIamPolicyRequest request = SetIamPolicyRequest.newBuilder()
677+
.setPolicy(PolicyMarshaller.INSTANCE.toPb(newPolicy))
678+
.setResource(SubscriberApi.formatSubscriptionName(options().projectId(), subscription))
679+
.build();
680+
return transform(rpc.setIamPolicy(request), POLICY_TO_PB_FUNCTION);
681+
}
682+
683+
@Override
684+
public List<Boolean> testSubscriptionPermissions(String subscription, List<String> permissions) {
685+
return get(testSubscriptionPermissionsAsync(subscription, permissions));
686+
}
687+
688+
@Override
689+
public Future<List<Boolean>> testSubscriptionPermissionsAsync(String subscription,
690+
List<String> permissions) {
691+
TestIamPermissionsRequest request = TestIamPermissionsRequest.newBuilder()
692+
.setResource(SubscriberApi.formatSubscriptionName(options().projectId(), subscription))
693+
.addAllPermissions(permissions)
694+
.build();
695+
return transform(rpc.testIamPermissions(request), permissionsFromPbFunction(permissions));
696+
}
697+
698+
private static Function<TestIamPermissionsResponse, List<Boolean>> permissionsFromPbFunction(
699+
final List<String> permissions) {
700+
return new Function<TestIamPermissionsResponse, List<Boolean>>() {
701+
@Override
702+
public List<Boolean> apply(TestIamPermissionsResponse response) {
703+
Set<String> permissionsOwned = ImmutableSet.copyOf(
704+
firstNonNull(response.getPermissionsList(), ImmutableList.<String>of()));
705+
ImmutableList.Builder<Boolean> answer = ImmutableList.builder();
706+
for (String permission : permissions) {
707+
answer.add(permissionsOwned.contains(permission));
708+
}
709+
return answer.build();
710+
}
711+
};
712+
}
713+
604714
static <T extends Option.OptionType> Map<Option.OptionType, ?> optionMap(Option... options) {
605715
Map<Option.OptionType, Object> optionMap = Maps.newHashMap();
606716
for (Option option : options) {

0 commit comments

Comments
 (0)