Micronaut HTTP Filter to modify the response payload

Question

I need to post process the response of any Micronaut controller and eliminate items in the response body when a user is not authorized to access them.

In a blocking world I would implement it like

protected MutableHttpResponse<?> doFilterOnce(HttpRequest<?> request, ServerFilterChain chain) {

        // If Micronaut Security rejected the request simpy do nothing
        if (request.getAttribute(SecurityFilter.REJECTION).isPresent()) {
            log.debug("Request was previously rejected. Not going to contact PDP");
            return chain.proceed(request);
        }

        HttpMethod method = request.getMethod();

        if (method.equals(GET) || method.equals(HEAD)) {

            MutableHttpResponse<?> response = chain.proceed(request);

            if (response.getBody().isPresent()) {
                // iterate through the body

                Object theBody = response.getBody().get();

                if (theBody instanceof Collection) {

                    Collection<?> iterable = (Iterable<?>) theBody;

                    // select all elements that are rejected. This is a blocking call.
                    List<?> collect = iterable.stream().filter(item -> mySecService.isAllowed(item) == false).collect(Collectors.toList());
                    // remove them
                    iterable.removeAll(collect);
                    // reset the body
                    response.body(iterable);
                }
            }

        } else {
            return chain.proceed(request)
        }
        return response;
    }

Micronaut states that

Filters execute in the event loop therefore blocking operations must be offloaded to another thread pool.

and therefore in the real world it requires mit to return

1. a Flowable
2. implement the code above in a reactive way

This is what I have done so far.

if (method.equals(GET) || method.equals(HEAD)) {
    // post process
    return Flowable.fromPublisher(chain.proceed(request))
            .doNext(response -> {
                Optional<?> body = response.getBody();

                if (body.isPresent()) {
                     // how can I continue here an process the response body collection?
                }
            });
}

Can someone give me a hint how to continue processing the response body, do the security check, remove the items and reset the new body?

Answer 1

In the reactive world, we will have to use Flowable in place of Stream. Given a reactive implementation of the method used to check if a certain item is allowed (e.g some method mySecService::isAllowedReactive that returns a Single<Boolean>), we can use Flowable#concatMapSingle to create a Flowable that tells us if the user can see n'th item. We can combine this with the original iterable to filter out disallowed items. And of course, nesting reactive types like this requires the use of flatMap.

@Override
protected Publisher<MutableHttpResponse<?>> doFilterOnce(HttpRequest<?> request, ServerFilterChain chain) {
    // If Micronaut Security rejected the request simpy do nothing
    if (request.getAttribute(SecurityFilter.REJECTION).isPresent()) {
        log.debug("Request was previously rejected. Not going to contact PDP");
        return chain.proceed(request);
    }
    HttpMethod method = request.getMethod();
    return Flowable.fromPublisher(chain.proceed(request)).flatMap(response -> {
        if (method.equals(HttpMethod.GET) || method.equals(HttpMethod.HEAD)) {
            if (response.getBody().isPresent()) {
                // iterate through the body
                Object theBody = response.getBody().get();
                if (theBody instanceof Collection) {
                    Collection<?> iterable = (Collection<?>) theBody;



                    // find which elements are not allowed.
                    Flowable<Boolean> isElementAllowed = Flowable.fromIterable(iterable)
                            .concatMapSingle(mySecService::isAllowedReactive);



                    // filter out forbidden elements
                    Single<ArrayList<? super Object>> allowedElementsSingle = isElementAllowed
                            .zipWith(iterable, (isAllowed, item) -> {
                                if (isAllowed) {
                                    return Optional.of(item);
                                } else {
                                    return Optional.empty();
                                }
                            })
                            .filter(Optional::isPresent)
                            .map(Optional::get)
                            .collect(ArrayList::new, ArrayList::add);



                    return allowedElementsSingle.flatMapPublisher(allowedElements -> {
                        // reset the body
                        response.body(allowedElements);
                        return Flowable.just(response);
                    });
                }
            }
        }
        return Flowable.just(response);
    });
}

Answer 2

i don't know if this is answer for your question but if you d'like to add something to response or to all responses, for example set response headers, you can use filter and then use response to add something, example below.

@Filter("/repos/**")
public class CustomFilter implements HttpClientFilter {

final String username;
final String token;

CustomFilter(
        @Value("${username}") String username,
        @Value("${token}") String token) {
    this.username = username;
    this.token = token;
}

@Override
public Publisher<MutableHttpResponse<?>> doFilter(HttpRequest<?> request, ServerFilterChain chain) {
    return Publishers.map(Flux.fromIterable(authenticationFetchers)
            .flatMap(authenticationFetcher -> authenticationFetcher.fetchAuthentication(request))
            .next(), this::setCustomHeaders);
}

private MutableHttpResponse<?> setCustomHeaders(MutableHttpResponse<?> response) {
    response.header(X_CONTENT_TYPE_KEY, X_CONTENT_TYPE_VALUE);
    return response;
    }
}