Skip to content

Centralize transport identity propagation #6477

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

Open
cwperks wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Centralize transport identity propagation #6477

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
108 changes: 2 additions & 106 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 @@ -53,20 +53,16 @@
import org.opensearch.common.settings.Settings;
import org.opensearch.common.util.concurrent.ThreadContext;
import org.opensearch.core.common.io.stream.StreamInput;
import org.opensearch.core.common.transport.TransportAddress;
import org.opensearch.core.transport.TransportResponse;
import org.opensearch.security.OpenSearchSecurityPlugin;
import org.opensearch.security.auditlog.AuditLog;
import org.opensearch.security.auditlog.AuditLog.Origin;
import org.opensearch.security.auth.BackendRegistry;
import org.opensearch.security.configuration.ClusterInfoHolder;
import org.opensearch.security.privileges.dlsfls.DlsFlsLegacyHeaders;
import org.opensearch.security.ssl.SslExceptionHandler;
import org.opensearch.security.ssl.transport.PrincipalExtractor;
import org.opensearch.security.ssl.transport.SSLConfig;
import org.opensearch.security.support.Base64Helper;
import org.opensearch.security.support.ConfigConstants;
import org.opensearch.security.user.User;
import org.opensearch.security.user.UserFactory;
import org.opensearch.tasks.Task;
import org.opensearch.threadpool.ThreadPool;
Expand Down Expand Up @@ -152,21 +148,15 @@ public <T extends TransportResponse> void sendRequestDecorate(
DiscoveryNode localNode
) {
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);
final String injectedRolesString = getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES);
final TransportIdentityContext identityContext = TransportIdentityContext.capture(getThreadContext());
final String injectedRolesValidationString = getThreadContext().getTransient(
ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES_VALIDATION
);
final String origin0 = getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_ORIGIN);
final Object remoteAddress0 = getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS);
final String origCCSTransientDls = getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_DLS_QUERY_CCS);
final String origCCSTransientFls = getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_FLS_FIELDS_CCS);
final String origCCSTransientMf = getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_MASKED_FIELD_CCS);
final DlsFlsLegacyHeaders dlsFlsLegacyHeaders = getThreadContext().getTransient(DlsFlsLegacyHeaders.TRANSIENT_HEADER);

final User authenticatedUser = (User) getThreadContext().getPersistent(ConfigConstants.OPENDISTRO_SECURITY_AUTHENTICATED_USER);

final boolean isDebugEnabled = log.isDebugEnabled();
final boolean isStreamChannel = options != null && TransportRequestOptions.Type.STREAM.equals(options.type());
// skip the same node optimization for stream transport which doesn't use DirectChannel and thus ser/de is needed
Expand Down Expand Up @@ -280,15 +270,7 @@ && getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROL

getThreadContext().putHeader(headerMap);

ensureCorrectHeaders(
remoteAddress0,
user0,
authenticatedUser,
origin0,
injectedUserString,
injectedRolesString,
isSameNodeRequest
);
identityContext.propagate(getThreadContext(), isSameNodeRequest);

if (actionTraceEnabled.get()) {
getThreadContext().putHeader(
Expand All @@ -313,92 +295,6 @@ boolean isCrossClusterSearchEnabled() {
return OpenSearchSecurityPlugin.GuiceHolder.getRemoteClusterService().isCrossClusterSearchEnabled();
}

private void ensureCorrectHeaders(
final Object remoteAdr,
final User origUser,
final User authenticatedUser,
final String origin,
final String injectedUserString,
final String injectedRolesString,
final boolean isSameNodeRequest
) {
// keep original address

if (origin != null
&& !origin.isEmpty() /*&& !Origin.LOCAL.toString().equalsIgnoreCase(origin)*/
&& getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_ORIGIN_HEADER) == null) {
getThreadContext().putHeader(ConfigConstants.OPENDISTRO_SECURITY_ORIGIN_HEADER, origin);
}

if (origin == null && getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_ORIGIN_HEADER) == null) {
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) {
transportAddress = (TransportAddress) remoteAdr;
}
}

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

if (origUser != null) {
// 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.
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_USER, origUser);
} else if (StringUtils.isNotEmpty(injectedRolesString)) {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES, injectedRolesString);
} else if (StringUtils.isNotEmpty(injectedUserString)) {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_USER, injectedUserString);
}
} else {
if (transportAddress != null) {
getThreadContext().putHeader(
ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS_HEADER,
Base64Helper.serializeObject(transportAddress.address())
);
}

// Propagate the authenticated user so it can be restored when the request is received.
String authenticatedUserHeader = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_AUTHENTICATED_USER_HEADER);
String userSameAsAuthenticatedUserHeader = getThreadContext().getHeader(
ConfigConstants.OPENDISTRO_SECURITY_USER_SAME_AS_SUBJECT_HEADER
);
if (authenticatedUserHeader == null && authenticatedUser != null) {
if (origUser != null && origUser.equals(authenticatedUser)) {
if (userSameAsAuthenticatedUserHeader == null) {
getThreadContext().putHeader(
ConfigConstants.OPENDISTRO_SECURITY_USER_SAME_AS_SUBJECT_HEADER,
Boolean.TRUE.toString()
);
}
} else {
getThreadContext().putHeader(
ConfigConstants.OPENDISTRO_SECURITY_AUTHENTICATED_USER_HEADER,
authenticatedUser.toSerializedBase64()
);
}
}
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, origUser.toSerializedBase64());
} 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() {
return threadPool.getThreadContext();
}
Expand Down
89 changes: 8 additions & 81 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 @@ -26,7 +26,6 @@

package org.opensearch.security.transport;

import java.net.InetSocketAddress;
import java.security.cert.X509Certificate;
import java.util.List;
import java.util.UUID;
Expand All @@ -40,7 +39,6 @@
import org.opensearch.action.support.replication.TransportReplicationAction.ConcreteShardRequest;
import org.opensearch.cluster.service.ClusterService;
import org.opensearch.common.util.concurrent.ThreadContext;
import org.opensearch.core.common.transport.TransportAddress;
import org.opensearch.extensions.ExtensionsManager;
import org.opensearch.search.internal.ShardSearchRequest;
import org.opensearch.security.DefaultObjectMapper;
Expand All @@ -53,10 +51,8 @@
import org.opensearch.security.ssl.transport.SSLConfig;
import org.opensearch.security.ssl.transport.SecuritySSLRequestHandler;
import org.opensearch.security.ssl.util.ExceptionUtils;
import org.opensearch.security.support.Base64Helper;
import org.opensearch.security.support.ConfigConstants;
import org.opensearch.security.support.HeaderHelper;
import org.opensearch.security.user.User;
import org.opensearch.security.user.UserFactory;
import org.opensearch.security.util.ParentChildrenQueryDetector;
import org.opensearch.tasks.Task;
Expand Down Expand Up @@ -121,11 +117,7 @@ protected void messageReceivedDecorate(

final ThreadContext.StoredContext sgContext = getThreadContext().newStoredContext(false);

final String originHeader = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_ORIGIN_HEADER);

if (!Strings.isNullOrEmpty(originHeader)) {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_ORIGIN, originHeader);
}
TransportIdentityContext.restoreOrigin(getThreadContext());

// restore headers used for DLS
final var dlsRequestHeadersAsString = getThreadContext().getHeader(ConfigConstants.OPENSEARCH_SECURITY_DLS_REQUEST_HEADERS);
Expand Down Expand Up @@ -162,17 +154,8 @@ protected void messageReceivedDecorate(
}

// bypass non-netty requests
if (getThreadContext().getTransient(ConfigConstants.OPENDISTRO_SECURITY_USER) != null
|| 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 (TransportIdentityContext.hasTransientIdentity(getThreadContext())) {
TransportIdentityContext.restoreRolesValidation(getThreadContext());

if (isActionTraceEnabled()) {
getThreadContext().putHeader(
Expand All @@ -187,68 +170,12 @@ protected void messageReceivedDecorate(

putInitialActionClassHeader(initialActionClassValue, resolvedActionClass);
} else {
String authenticatedUserHeader = getThreadContext().getHeader(
ConfigConstants.OPENDISTRO_SECURITY_AUTHENTICATED_USER_HEADER
);
String userSameAsAuthenticatedUserHeader = getThreadContext().getHeader(
ConfigConstants.OPENDISTRO_SECURITY_USER_SAME_AS_SUBJECT_HEADER
);
String userHeader = getThreadContext().getHeader(ConfigConstants.OPENDISTRO_SECURITY_USER_HEADER);

// Deserialize and sanitize users.
User user = null;
if (userHeader != null) {
user = this.userFactory.fromSerializedBase64(userHeader);
user = remoteClusterIdentityPolicy.sanitize(user, getThreadContext());
}
User authenticatedUser = null;
if (authenticatedUserHeader != null) {
authenticatedUser = this.userFactory.fromSerializedBase64(authenticatedUserHeader);
authenticatedUser = remoteClusterIdentityPolicy.sanitize(authenticatedUser, getThreadContext());
}

// Store the authenticated user in persistent context (if not already set).
if (getThreadContext().getPersistent(ConfigConstants.OPENDISTRO_SECURITY_AUTHENTICATED_USER) == null) {
if (Boolean.parseBoolean(userSameAsAuthenticatedUserHeader) && user != null) {
getThreadContext().putPersistent(ConfigConstants.OPENDISTRO_SECURITY_AUTHENTICATED_USER, user);
} else if (authenticatedUser != null) {
getThreadContext().putPersistent(ConfigConstants.OPENDISTRO_SECURITY_AUTHENTICATED_USER, authenticatedUser);
}
}

// Store transient user or injected roles
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, user);
}

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
TransportIdentityContext.restoreSerializedIdentity(
getThreadContext(),
request.remoteAddress(),
userFactory,
remoteClusterIdentityPolicy
);
if (!Strings.isNullOrEmpty(rolesValidation)) {
getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES_VALIDATION, rolesValidation);
}
}

if (channelType.equals("direct")) {
Expand Down
Loading
Loading