Skip to content

Making Pulsar Proxy more secure #1002

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jai1 merged 31 commits into from
Jan 31, 2018
Merged

Making Pulsar Proxy more secure #1002

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
31 commits
Select commit Hold shift + click to select a range
f956e8d
Made service discovery optional for proxy
Nov 23, 2017
878266a
Creating a proto file remains
Nov 28, 2017
85206bf
Updated proto file for Pulsar Proxy
Dec 4, 2017
d982ec7
Fixed compilation error
Dec 4, 2017
cee5751
Merge branch 'Proxy2' of https://github.com/jai1/pulsar into Proxy2
Dec 4, 2017
09d5bcf
Added tests and fixed some logic which causes existing test to break
Dec 8, 2017
0f77bfe
Fixed tests
Dec 15, 2017
069003c
Added certificates
Dec 19, 2017
76497c7
Tests
Dec 19, 2017
2516038
Tests
Dec 19, 2017
578fdb6
Readded Server connection code
Dec 20, 2017
9939742
Corrected and fixed some test cases
Dec 20, 2017
b716ae3
Removed sending of original auth data
Dec 21, 2017
bc95361
Handled merge conflict
Dec 21, 2017
25fe708
Fixed some spaces in license header and Commands.java
Dec 21, 2017
bc54ffd
Fixed Redirect logic
Dec 29, 2017
2071d13
Handled merge conflict
Dec 29, 2017
9ee13dd
Added licene headers
Dec 29, 2017
f5f7a2e
Resolved Merge Conflict
Jan 11, 2018
fd49d55
Corrected some logging and Start up dependencies
Jan 11, 2018
252cce3
Fixed test case dependencies
Jan 11, 2018
65b9cd5
Resolved merge conflict
Jan 15, 2018
b03cc36
Addressed Rajan's Comments
Jan 16, 2018
569bc8a
Addressed Matteo's comments
Jan 22, 2018
5656d6a
Simplified logic in AuthorizationManager.canLookupAsync as per Andrew…
Jan 26, 2018
a5c9b6a
Addressed Rajans changes
Jan 26, 2018
d50a8bf
Addressed Matteos comments
Jan 27, 2018
75958af
Fixed compilation errors
Jan 28, 2018
006e8c6
Changd cluster name to get tests to work
Jan 29, 2018
2bbac2f
Merge branch 'master' of https://github.com/yahoo/pulsar into Proxy2
Jan 29, 2018
5e5a2a2
Fixed test cases to not create clusters since broker already does
Jan 29, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 56 additions & 16 deletions ...ker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.pulsar.zookeeper.ZooKeeperCache.cacheTimeOutInSec;
Expand Down Expand Up @@ -66,8 +69,8 @@ public boolean canProduce(DestinationName destination, String role) throws Excep
log.warn("Time-out {} sec while checking authorization on {} ", cacheTimeOutInSec, destination);
throw e;
} catch (Exception e) {
log.warn("Producer-client with Role - {} failed to get permissions for destination - {}", role,
destination, e);
log.warn("Producer-client with Role - {} failed to get permissions for destination - {}. {}", role,
destination, e.getMessage());
throw e;
}
}
Expand Down Expand Up @@ -96,8 +99,9 @@ public CompletableFuture<Boolean> canConsumeAsync(DestinationName destination, S
switch (policies.get().subscription_auth_mode) {
case Prefix:
if (!subscription.startsWith(role)) {
PulsarServerException ex = new PulsarServerException(
String.format("Failed to create consumer - The subscription name needs to be prefixed by the authentication role, like %s-xxxx for destination: %s", role, destination));
PulsarServerException ex = new PulsarServerException(String.format(
"Failed to create consumer - The subscription name needs to be prefixed by the authentication role, like %s-xxxx for destination: %s",
role, destination));
permissionFuture.completeExceptionally(ex);
return;
}
Expand All @@ -111,13 +115,12 @@ public CompletableFuture<Boolean> canConsumeAsync(DestinationName destination, S
permissionFuture.complete(isAuthorized);
});
}).exceptionally(ex -> {
log.warn("Client with Role - {} failed to get permissions for destination - {}", role, destination,
ex);
log.warn("Client with Role - {} failed to get permissions for destination - {}. {}", role, destination, ex.getMessage());
permissionFuture.completeExceptionally(ex);
return null;
});
} catch (Exception e) {
log.warn("Client with Role - {} failed to get permissions for destination - {}", role, destination, e);
log.warn("Client with Role - {} failed to get permissions for destination - {}. {}", role, destination, e.getMessage());
permissionFuture.completeExceptionally(e);
}
return permissionFuture;
Expand All @@ -130,8 +133,8 @@ public boolean canConsume(DestinationName destination, String role, String subsc
log.warn("Time-out {} sec while checking authorization on {} ", cacheTimeOutInSec, destination);
throw e;
} catch (Exception e) {
log.warn("Consumer-client with Role - {} failed to get permissions for destination - {}", role,
destination, e);
log.warn("Consumer-client with Role - {} failed to get permissions for destination - {}. {}", role,
destination, e.getMessage());
throw e;
}
}
Expand All @@ -150,8 +153,46 @@ public boolean canLookup(DestinationName destination, String role) throws Except
return canProduce(destination, role) || canConsume(destination, role, null);
}

private CompletableFuture<Boolean> checkAuthorization(DestinationName destination, String role,
AuthAction action) {
/**
* Check whether the specified role can perform a lookup for the specified destination.
*
* For that the caller needs to have producer or consumer permission.
*
* @param destination
* @param role
* @return
* @throws Exception
*/
public CompletableFuture<Boolean> canLookupAsync(DestinationName destination, String role) {
CompletableFuture<Boolean> finalResult = new CompletableFuture<Boolean>();
canProduceAsync(destination, role).whenComplete((produceAuthorized, ex) -> {
if (ex == null) {
if (produceAuthorized) {
finalResult.complete(produceAuthorized);
return;
}
} else if (log.isDebugEnabled()) {

@rdhabalia rdhabalia Jan 30, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

else {
  if (log.isDebugEnabled()) {
    :
  }
}

log.debug("Destination [{}] Role [{}] exception occured while trying to check Produce permissions. {}",
destination.toString(), role, ex.getMessage());
}
canConsumeAsync(destination, role, null).whenComplete((consumeAuthorized, e) -> {
if (e == null) {
if (consumeAuthorized) {
finalResult.complete(consumeAuthorized);
return;
}
} else if (log.isDebugEnabled()) {
log.debug(

@rdhabalia rdhabalia Jan 30, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

canProduceAsync() and canConsumeAsync() fails the future when it has zk-connection issue and in that case, we should give 5XX error instead 403. So, I think we should fail finalResult future here instead returning completing it with false.

"Destination [{}] Role [{}] exception occured while trying to check Consume permissions. {}",
destination.toString(), role, e.getMessage());
}
finalResult.complete(false);
});
});
return finalResult;
}

private CompletableFuture<Boolean> checkAuthorization(DestinationName destination, String role, AuthAction action) {
if (isSuperUser(role)) {
return CompletableFuture.completedFuture(true);
} else {
Expand Down Expand Up @@ -218,13 +259,13 @@ public CompletableFuture<Boolean> checkPermission(DestinationName destination, S
}
permissionFuture.complete(false);
}).exceptionally(ex -> {
log.warn("Client with Role - {} failed to get permissions for destination - {}", role, destination,
ex);
log.warn("Client with Role - {} failed to get permissions for destination - {}. {}", role, destination,
ex.getMessage());
permissionFuture.completeExceptionally(ex);
return null;
});
} catch (Exception e) {
log.warn("Client with Role - {} failed to get permissions for destination - {}", role, destination, e);
log.warn("Client with Role - {} failed to get permissions for destination - {}. {}", role, destination, e.getMessage());
permissionFuture.completeExceptionally(e);
}
return permissionFuture;
Expand All @@ -244,8 +285,7 @@ private boolean checkWildcardPermission(String checkedRole, AuthAction checkedAc
}

// Suffix match
if (permittedRole.charAt(0) == '*'
&& checkedRole.endsWith(permittedRole.substring(1))
if (permittedRole.charAt(0) == '*' && checkedRole.endsWith(permittedRole.substring(1))
&& permittedActions.contains(checkedAction)) {
return true;
}
Expand Down
2 changes: 1 addition & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/PersistentTopics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1314,7 +1314,7 @@ public static CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMet
validateAdminAccessOnProperty(pulsar, clientAppId, dn.getProperty());
} catch (RestException authException) {
log.warn("Failed to authorize {} on cluster {}", clientAppId, dn.toString());
throw new PulsarClientException(String.format("Authorization failed %s on cluster %s with error %s",
throw new PulsarClientException(String.format("Authorization failed %s on topic %s with error %s",
clientAppId, dn.toString(), authException.getMessage()));
}
} catch (Exception ex) {
Expand Down
Loading