Set permissions/authentication for spring-cloud-stream message consumer so it passes @PreAuthorize checks

Question

I consume messages from spring-cloud-stream through a Consumer<MyMessage> Implementation. As part of the message handling I need to access methods that are protected with @PreAuthorize security-checks. By default the Consumer run unauthenticated so message-handling fails.

Consumer:

@Bean
public Consumer<MyMessage> exampleMessageConsumer(MyMessageConsumer consumer) {
    return consumer::handleMessage;
}

Secured Method:

@PreAuthorize("hasAuthority('ROLE_ADMIN') or hasAuthority('ROLE_USER')")
public void doSomething() { ... }

I dont just want to bypass security, so what is the easiest way to authenticate my Consumer so it passes the check?

EDIT: we are using google pubsub as a binder

Answer 1

I found two possible solutions to my problem

1. use springs RunAs support (baeldung) to add permissions to a security context for a specific method. If i do this i need to add ROLE_RUN_AS_USER to my secured methods. At scale this would complicated annotations a lot.

2. Manually change the security context before executing the handler method and return it to its original state afterwards.

I went with the second option. I would have liked a transparent solution but there does not appear to be one.

To make this work i created a class that wraps a functional interface with the changing code and returns it.

public class RunAs {

    @FunctionalInterface
    public interface RunAsMethod {
        void runWithException() throws Throwable;
    }

    public static <T> Consumer<T> createWriteConsumer(Consumer<T> originalConsumer) {
        return message -> runWithWritePermission(() -> originalConsumer.accept(message));
    }

    public static void runWithWritePermission(final RunAsMethod func) {
        final Authentication originalAuthentication = SecurityContextHolder.getContext().getAuthentication();

        final AnonymousAuthenticationToken token = new AnonymousAuthenticationToken(
            "system",
            originalAuthentication != null ? originalAuthentication.getPrincipal() : "system",
            AuthorityUtils.createAuthorityList("ROLE_ADMIN", "SCOPE_write")
        );

        SecurityContextHolder.getContext().setAuthentication(token);
        try {
            func.runWithException();
        } catch (Throwable e) {
            throw new RuntimeException("exception during method with altered permissions", e);
        } finally {
            SecurityContextHolder.getContext().setAuthentication(originalAuthentication);
        }
    }
}

Answer 2

For the Kafka binder:

Add an @EventListener to listen for ConsumerStartedEvents; you can then add the authentication to the security context via the SecurityContextHolder; this binds it to the thread; the same thread is used to call the listener.