Skip to content
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

Cleanup Listener Handling in AuthorizationService #75252

Merged
merged 4 commits into from
Jul 13, 2021
Merged

Cleanup Listener Handling in AuthorizationService #75252

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
dde1d11
Cleanup Listener Handling in AuthorizationService
original-brownbear Jul 12, 2021
935321d
fix format
original-brownbear Jul 12, 2021
97741c2
Merge remote-tracking branch 'elastic/master' into simplify-async-sup…
original-brownbear Jul 13, 2021
6b5b059
CR: rename
original-brownbear Jul 13, 2021
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
88 changes: 46 additions & 42 deletions ...n/security/src/main/java/org/elasticsearch/xpack/security/authz/AuthorizationService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@
import org.elasticsearch.ElasticsearchSecurityException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.StepListener;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesAction;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
Expand All @@ -27,6 +26,7 @@
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.metadata.Metadata;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.util.concurrent.ListenableFuture;
import org.elasticsearch.core.Nullable;
import org.elasticsearch.common.Strings;
import org.elasticsearch.core.Tuple;
Expand Down Expand Up @@ -285,18 +285,24 @@ private void authorizeAction(final RequestInfo requestInfo, final String request
final AsyncSupplier<Set<String>> authorizedIndicesSupplier = new CachingAsyncSupplier<>(authzIndicesListener ->
authzEngine.loadAuthorizedIndices(requestInfo, authzInfo, metadata.getIndicesLookup(),
authzIndicesListener));
final AsyncSupplier<ResolvedIndices> resolvedIndicesAsyncSupplier = new CachingAsyncSupplier<>((resolvedIndicesListener) -> {
authorizedIndicesSupplier.getAsync(ActionListener.wrap(authorizedIndices -> {
resolveIndexNames(action, request, metadata, authorizedIndices, resolvedIndicesListener);
}, e -> {
auditTrail.accessDenied(requestId, authentication, action, request, authzInfo);
if (e instanceof IndexNotFoundException) {
listener.onFailure(e);
} else {
listener.onFailure(denialException(authentication, action, request, e));
}
}));
});
final AsyncSupplier<ResolvedIndices> resolvedIndicesAsyncSupplier = new CachingAsyncSupplier<>(resolvedIndicesListener ->
authorizedIndicesSupplier.getAsync(
ActionListener.wrap(
authorizedIndices ->
resolvedIndicesListener.onResponse(
Copy link
Member Author

Choose a reason for hiding this comment

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

inlined this to make the stack-trace one less deep and generally make the logic easier to follow, no need to have a method to resolve a listener when the method is blocking and one line anyway :)

indicesAndAliasesResolver.resolve(action, request, metadata, authorizedIndices)
),
e -> {
auditTrail.accessDenied(requestId, authentication, action, request, authzInfo);
if (e instanceof IndexNotFoundException) {
listener.onFailure(e);
} else {
listener.onFailure(denialException(authentication, action, request, e));
}
}
)
)
);
authzEngine.authorizeIndexAction(requestInfo, authzInfo, resolvedIndicesAsyncSupplier,
metadata.getIndicesLookup(), wrapPreservingContext(new AuthorizationResultListener<>(result ->
handleIndexActionAuthorizationResult(result, requestInfo, requestId, authzInfo, authzEngine, authorizedIndicesSupplier,
Expand Down Expand Up @@ -368,21 +374,19 @@ private void runRequestInterceptors(RequestInfo requestInfo, AuthorizationInfo a
if (requestInterceptors.isEmpty()) {
listener.onResponse(null);
} else {
Iterator<RequestInterceptor> requestInterceptorIterator = requestInterceptors.iterator();
final StepListener<Void> firstStepListener = new StepListener<>();
final RequestInterceptor first = requestInterceptorIterator.next();

StepListener<Void> prevListener = firstStepListener;
while (requestInterceptorIterator.hasNext()) {
final RequestInterceptor nextInterceptor = requestInterceptorIterator.next();
final StepListener<Void> current = new StepListener<>();
prevListener.whenComplete(v -> nextInterceptor.intercept(requestInfo, authorizationEngine, authorizationInfo, current),
listener::onFailure);
prevListener = current;
}

prevListener.addListener(listener);
first.intercept(requestInfo, authorizationEngine, authorizationInfo, firstStepListener);
final Iterator<RequestInterceptor> requestInterceptorIterator = requestInterceptors.iterator();
requestInterceptorIterator.next().intercept(requestInfo, authorizationEngine, authorizationInfo,
new ActionListener.Delegating<>(listener) {
@Override
public void onResponse(Void unused) {
if (requestInterceptorIterator.hasNext()) {
requestInterceptorIterator.next().intercept(requestInfo, authorizationEngine, authorizationInfo, this);
} else {
listener.onResponse(null);
}
}
}
);
Comment on lines +377 to +389
Copy link
Member

Choose a reason for hiding this comment

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

👍

}
}

Expand Down Expand Up @@ -594,11 +598,6 @@ private static String getAction(BulkItemRequest item) {
throw new IllegalArgumentException("No equivalent action for opType [" + docWriteRequest.opType() + "]");
}

private void resolveIndexNames(String action, TransportRequest request, Metadata metadata, Set<String> authorizedIndices,
ActionListener<ResolvedIndices> listener) {
listener.onResponse(indicesAndAliasesResolver.resolve(action, request, metadata, authorizedIndices));
}

private void putTransientIfNonExisting(String key, Object value) {
Object existing = threadContext.getTransient(key);
if (existing == null) {
Expand Down Expand Up @@ -712,22 +711,27 @@ private void handleFailure(boolean audit, @Nullable String context, @Nullable Ex
private static class CachingAsyncSupplier<V> implements AsyncSupplier<V> {

private final AsyncSupplier<V> asyncSupplier;
private V value = null;
private volatile ListenableFuture<V> valueFuture = null;

private CachingAsyncSupplier(AsyncSupplier<V> supplier) {
this.asyncSupplier = supplier;
}

@Override
public synchronized void getAsync(ActionListener<V> listener) {
if (value == null) {
asyncSupplier.getAsync(ActionListener.wrap(loaded -> {
value = loaded;
listener.onResponse(value);
}, listener::onFailure));
} else {
listener.onResponse(value);
public void getAsync(ActionListener<V> listener) {
if (valueFuture == null) {
boolean firstInvocation = false;
synchronized (this) {
if (valueFuture == null) {
valueFuture = new ListenableFuture<>();
firstInvocation = true;
}
}
if (firstInvocation) {
asyncSupplier.getAsync(valueFuture);
}
}
valueFuture.addListener(listener);
}
}

Expand Down