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

Adds a check to skip serialization-deserialization if request is for same node #2765

Merged
merged 23 commits into from
Jul 10, 2023
Merged

Adds a check to skip serialization-deserialization if request is for same node #2765

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
b76e804
Adds a check to skip serialization-deserialization if user request is…
DarshitChanpura May 11, 2023
4a8dda5
Fixes unused import
DarshitChanpura May 11, 2023
2e693cf
Simplifies direct request check
DarshitChanpura May 11, 2023
25796cb
Cleans up debug code
DarshitChanpura May 11, 2023
d8a4f9c
Changes direct request comparision to compare using localNode
DarshitChanpura May 12, 2023
44d5048
Addresses PR comments
DarshitChanpura May 12, 2023
d60bbb2
Cleans up logic a little
DarshitChanpura May 15, 2023
a55a0a9
Graceful error handling
DarshitChanpura May 16, 2023
73ff9cb
Adds a null check to avoid rewriting persistent header
DarshitChanpura May 16, 2023
62b6172
Changes same node evaluation logic to use TransportService instead of…
DarshitChanpura May 23, 2023
8a09ea8
Fetches local node from TransportService instance via GuiceHolder
DarshitChanpura May 24, 2023
fbb6041
Fixes spotless errors
DarshitChanpura May 24, 2023
ab7177d
Updates onNodeStarted to use the overloaded definition to accept loca…
DarshitChanpura Jun 6, 2023
4a0125b
Merge branch 'main' into perf-bug-2724
DarshitChanpura Jun 21, 2023
79ad17f
Adds security interceptor tests
DarshitChanpura Jun 28, 2023
20af819
Fixes checkstyle errors
DarshitChanpura Jun 28, 2023
0d4aa13
Merge remote-tracking branch 'upstream/main' into perf-bug-2724
DarshitChanpura Jun 29, 2023
2ae647a
Merge branch 'main' into perf-bug-2724
DarshitChanpura Jul 5, 2023
fc1234d
Check for transient headers in received
cwperks Jun 30, 2023
f114b72
Attempt to make logic simpler by checking for transients or request h…
cwperks Jun 30, 2023
35de56d
Remove add tc header
cwperks Jun 30, 2023
219b0b3
Handle skipSecurityInDualMode
cwperks Jun 30, 2023
691019b
Merge pull request #6 from cwperks/perf-bug-2724-craig
DarshitChanpura Jul 5, 2023
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
13 changes: 12 additions & 1 deletion src/main/java/org/opensearch/security/OpenSearchSecurityPlugin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@
import org.opensearch.action.support.ActionFilter;
import org.opensearch.client.Client;
import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
import org.opensearch.cluster.node.DiscoveryNode;
import org.opensearch.cluster.node.DiscoveryNodes;
import org.opensearch.cluster.service.ClusterService;
import org.opensearch.common.component.Lifecycle.State;
Expand Down Expand Up @@ -209,6 +210,7 @@ public final class OpenSearchSecurityPlugin extends OpenSearchSecuritySSLPlugin
private volatile ConfigurationRepository cr;
private volatile AdminDNs adminDns;
private volatile ClusterService cs;
private static volatile DiscoveryNode localNode;
private volatile AuditLog auditLog;
private volatile BackendRegistry backendRegistry;
private volatile SslExceptionHandler sslExceptionHandler;
Expand Down Expand Up @@ -1792,11 +1794,12 @@ public List<String> getSettingsFilter() {
}

@Override
public void onNodeStarted() {
public void onNodeStarted(DiscoveryNode localNode) {
log.info("Node started");
if (!SSLConfig.isSslOnlyMode() && !client && !disabled) {
cr.initOnNodeStart();
}
this.localNode = localNode;
final Set<ModuleInfo> securityModules = ReflectionHelper.getModulesLoaded();
log.info("{} OpenSearch Security modules loaded so far: {}", securityModules.size(), securityModules);
}
Expand Down Expand Up @@ -1876,6 +1879,14 @@ private static String handleKeyword(final String field) {
return field;
}

public static DiscoveryNode getLocalNode() {
return localNode;
}

public static void setLocalNode(DiscoveryNode node) {
Copy link
Member Author

Choose a reason for hiding this comment

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

need this for security Interceptor tests

Copy link
Member

Choose a reason for hiding this comment

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

Is it possible to add localNode to SecurityInterceptor constructor and remove this static method only needed for testing?

localNode = node;
}

public static class GuiceHolder implements LifecycleComponent {

private static RepositoriesService repositoriesService;
Expand Down
53 changes: 38 additions & 15 deletions src/main/java/org/opensearch/security/transport/SecurityInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
import org.opensearch.action.get.GetRequest;
import org.opensearch.action.search.SearchAction;
import org.opensearch.action.search.SearchRequest;
import org.opensearch.cluster.node.DiscoveryNode;
import org.opensearch.cluster.service.ClusterService;
import org.opensearch.common.io.stream.StreamInput;
import org.opensearch.common.settings.Settings;
Expand Down Expand Up @@ -131,7 +132,6 @@ public <T extends TransportResponse> void sendRequestDecorate(
TransportRequestOptions options,
TransportResponseHandler<T> handler
) {

final Map<String, String> origHeaders0 = getThreadContext().getHeaders();
final User user0 = getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_USER);
final String injectedUserString = getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_USER);
Expand All @@ -146,6 +146,9 @@ public <T extends TransportResponse> void sendRequestDecorate(
final String origCCSTransientMf = getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_MASKED_FIELD_CCS);

final boolean isDebugEnabled = log.isDebugEnabled();
final DiscoveryNode localNode = OpenSearchSecurityPlugin.getLocalNode();
boolean isSameNodeRequest = localNode != null && localNode.equals(connection.getNode());

try (ThreadContext.StoredContext stashedContext = getThreadContext().stashContext()) {
Copy link
Member Author

@DarshitChanpura DarshitChanpura May 11, 2023 •
edited
Loading

Choose a reason for hiding this comment

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

Could not yet find a solution to not stash the context for same node.

Tried with these changes:

try (ThreadContext.StoredContext stashedContext = (isDirectRequest ? null: getThreadContext().stashContext())) {
    final TransportResponseHandler<T> restoringHandler = isDirectRequest ? handler : new RestoringTransportResponseHandler<T>(handler, stashedContext);

But the test fails with ClusterManagernotDiscoveredException everytime on test start.

Copy link
Contributor

Choose a reason for hiding this comment

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

It is perhaps good idea to continue stashing from security point of view too. That way, any existing transients assumed to be cleared are not carry forwarded.

For example let us assume current thread context had a,b,c entries and we explicitly passed around only a & b. Now carry forwarding c may be a security concern.

Copy link
Collaborator

Choose a reason for hiding this comment

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

But the test fails with ClusterManagernotDiscoveredException everytime on test start.

I would guess that this is related to the wrong check for isDirectRequest. But, yes, I would also recommend to stash anyway and just restore the transients which you want to keep. That would be the most conservative and most minimal change.

Copy link
Member

Choose a reason for hiding this comment

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

should we say isLocalNodeRequest to add more readability? or are we categorizing local node request to direct channel through out the code-base ?

final TransportResponseHandler<T> restoringHandler = new RestoringTransportResponseHandler<T>(handler, stashedContext);
getThreadContext().putHeader("_opendistro_security_remotecn", cs.getClusterName().value());
Expand Down Expand Up @@ -223,7 +226,7 @@ && getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROL

getThreadContext().putHeader(headerMap);

ensureCorrectHeaders(remoteAddress0, user0, origin0, injectedUserString, injectedRolesString);
ensureCorrectHeaders(remoteAddress0, user0, origin0, injectedUserString, injectedRolesString, isSameNodeRequest);

if (isActionTraceEnabled()) {
getThreadContext().putHeader(
Expand All @@ -249,7 +252,8 @@ private void ensureCorrectHeaders(
final User origUser,
final String origin,
final String injectedUserString,
final String injectedRolesString
final String injectedRolesString,
boolean isSameNodeRequest
) {
// keep original address

Expand All @@ -263,30 +267,49 @@ && getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_ORIGIN_HEADE
getThreadContext().putHeader(ConfigConstants.OPENDISTRO_SECURITY_ORIGIN_HEADER, Origin.LOCAL.toString());
}

TransportAddress transportAddress = null;
if (remoteAdr != null && remoteAdr instanceof TransportAddress) {

String remoteAddressHeader = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS_HEADER);

if (remoteAddressHeader == null) {
getThreadContext().putHeader(
ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS_HEADER,
Base64Helper.serializeObject(((TransportAddress) remoteAdr).address())
);
transportAddress = (TransportAddress) remoteAdr;
}
}

String userHeader = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_USER_HEADER);
// we put headers as transient for same node requests
if (isSameNodeRequest) {
if (transportAddress != null) {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS, transportAddress);
}

if (userHeader == null) {
if (origUser != null) {
getThreadContext().putHeader(ConfigConstants.OPENDISTRO_SECURITY_USER_HEADER, Base64Helper.serializeObject(origUser));
// if request is going to be handled by same node, we directly put transient value as the thread context is not going to be
// stah.
Copy link
Contributor

Choose a reason for hiding this comment

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

Suggested change
// stah.
// stashed.

getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_USER, origUser);
} else if (StringUtils.isNotEmpty(injectedRolesString)) {
getThreadContext().putHeader(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES_HEADER, injectedRolesString);
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES, injectedRolesString);
} else if (StringUtils.isNotEmpty(injectedUserString)) {
getThreadContext().putHeader(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_USER_HEADER, injectedUserString);
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_USER, injectedUserString);
}
} else {
Copy link
Contributor

Choose a reason for hiding this comment

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

Do we not intend to retain the existing userHeader == null check for this else code block, or are we assuming that this scenario used to arise only for direct requests?

if (transportAddress != null) {
getThreadContext().putHeader(
ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS_HEADER,
Base64Helper.serializeObject(transportAddress.address())
);
}
}

final String userHeader = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_USER_HEADER);
if (userHeader == null) {
// put as headers for other requests
if (origUser != null) {
getThreadContext().putHeader(ConfigConstants.OPENDISTRO_SECURITY_USER_HEADER, Base64Helper.serializeObject(origUser));
} else if (StringUtils.isNotEmpty(injectedRolesString)) {
getThreadContext().putHeader(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES_HEADER, injectedRolesString);
} else if (StringUtils.isNotEmpty(injectedUserString)) {
getThreadContext().putHeader(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_USER_HEADER, injectedUserString);
}
}
}
}

private ThreadContext getThreadContext() {
Expand Down
103 changes: 33 additions & 70 deletions src/main/java/org/opensearch/security/transport/SecurityRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,6 @@ protected void messageReceivedDecorate(
final TransportChannel transportChannel,
Task task
) throws Exception {

String resolvedActionClass = request.getClass().getSimpleName();

if (request instanceof BulkShardRequest) {
Expand Down Expand Up @@ -142,7 +141,31 @@ protected void messageReceivedDecorate(
}

// bypass non-netty requests
if (channelType.equals("direct")) {
if (getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_USER) != null
Copy link
Member

Choose a reason for hiding this comment

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

This logic in the receiver was simplified using @krishna-ggk 's suggestion here: #2765 (comment)

The receiver does not need any logic to determine where the request came from, if the transient headers are present then there is no need to deserialize. If the serialized (non-transient) headers are present then it will deserialize them.

|| getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_USER) != null
|| getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES) != null
|| getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS) != null) {

final String rolesValidation = getThreadContext().getHeader(
ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES_VALIDATION_HEADER
);
if (!Strings.isNullOrEmpty(rolesValidation)) {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES_VALIDATION, rolesValidation);
}

if (isActionTraceEnabled()) {
getThreadContext().putHeader(
"_opendistro_security_trace" + System.currentTimeMillis() + "#" + UUID.randomUUID().toString(),
Thread.currentThread().getName()
+ " DIR -> "
+ transportChannel.getChannelType()
+ " "
+ getThreadContext().getHeaders()
);
}

putInitialActionClassHeader(initialActionClassValue, resolvedActionClass);
} else {
final String userHeader = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_USER_HEADER);
final String injectedRolesHeader = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES_HEADER);
final String injectedUserHeader = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_USER_HEADER);
Expand All @@ -162,15 +185,15 @@ protected void messageReceivedDecorate(
);
}

final String originalRemoteAddress = getThreadContext().getHeader(
ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS_HEADER
);
String originalRemoteAddress = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS_HEADER);

if (!Strings.isNullOrEmpty(originalRemoteAddress)) {
getThreadContext().putTransient(
ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS,
new TransportAddress((InetSocketAddress) Base64Helper.deserializeObject(originalRemoteAddress))
);
} else {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS, request.remoteAddress());
}

final String rolesValidation = getThreadContext().getHeader(
Expand All @@ -179,20 +202,9 @@ protected void messageReceivedDecorate(
if (!Strings.isNullOrEmpty(rolesValidation)) {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES_VALIDATION, rolesValidation);
}
}

if (isActionTraceEnabled()) {
getThreadContext().putHeader(
"_opendistro_security_trace" + System.currentTimeMillis() + "#" + UUID.randomUUID().toString(),
Thread.currentThread().getName()
+ " DIR -> "
+ transportChannel.getChannelType()
+ " "
+ getThreadContext().getHeaders()
);
}

putInitialActionClassHeader(initialActionClassValue, resolvedActionClass);

if (channelType.equals("direct")) {
super.messageReceivedDecorate(request, handler, transportChannel, task);
return;
}
Expand Down Expand Up @@ -272,58 +284,10 @@ protected void messageReceivedDecorate(

// network intercluster request or cross search cluster request
// CS-SUPPRESS-SINGLE: RegexpSingleline Used to allow/disallow TLS connections to extensions
if (HeaderHelper.isInterClusterRequest(getThreadContext())
if (!(HeaderHelper.isInterClusterRequest(getThreadContext())
Copy link
Contributor

Choose a reason for hiding this comment

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

We either need to update the comment above this or remove the exclamation point I think.

|| HeaderHelper.isTrustedClusterRequest(getThreadContext())
|| HeaderHelper.isExtensionRequest(getThreadContext())) {
|| HeaderHelper.isExtensionRequest(getThreadContext()))) {
// CS-ENFORCE-SINGLE

final String userHeader = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_USER_HEADER);
final String injectedRolesHeader = getThreadContext().getHeader(
ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES_HEADER
);
final String injectedUserHeader = getThreadContext().getHeader(
ConfigConstants.OPENDISTRO_SECURITY_INJECTED_USER_HEADER
);

if (Strings.isNullOrEmpty(userHeader)) {
// Keeping role injection with higher priority as plugins under OpenSearch will be using this
// on transport layer
if (!Strings.isNullOrEmpty(injectedRolesHeader)) {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES, injectedRolesHeader);
} else if (!Strings.isNullOrEmpty(injectedUserHeader)) {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_USER, injectedUserHeader);
}
} else {
getThreadContext().putTransient(
ConfigConstants.OPENDISTRO_SECURITY_USER,
Objects.requireNonNull((User) Base64Helper.deserializeObject(userHeader))
);
}

String originalRemoteAddress = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS_HEADER);

if (!Strings.isNullOrEmpty(originalRemoteAddress)) {
getThreadContext().putTransient(
ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS,
new TransportAddress((InetSocketAddress) Base64Helper.deserializeObject(originalRemoteAddress))
);
} else {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS, request.remoteAddress());
}

final String rolesValidation = getThreadContext().getHeader(
ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES_VALIDATION_HEADER
);
if (!Strings.isNullOrEmpty(rolesValidation)) {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES_VALIDATION, rolesValidation);
}

} else {
// this is a netty request from a non-server node (maybe also be internal: or a shard request)
// and therefore issued by a transport client

// since OS 2.0 we do not support this any longer because transport client no longer available

final OpenSearchException exception = ExceptionUtils.createTransportClientNoLongerSupportedException();
log.error(exception.toString());
transportChannel.sendResponse(exception);
Expand All @@ -346,9 +310,8 @@ protected void messageReceivedDecorate(
}

putInitialActionClassHeader(initialActionClassValue, resolvedActionClass);

super.messageReceivedDecorate(request, handler, transportChannel, task);
}
super.messageReceivedDecorate(request, handler, transportChannel, task);
} finally {

if (isActionTraceEnabled()) {
Expand Down
Loading