MQTT topics with path parameters required to match guarded identity

[epieffe]

The goal of this PR is to allow secure access to some MQTT client specific topics for publish and/or subscribe.

For publish or subscribe routes:

• mqtt server can define path parameters for any segment in a topic name
• mqtt server can enforce that a path parameter matches the client's guarded identity (e.g. via jwt guard)

Example configuration:

mqtt_server0:
  type: mqtt
  kind: server
  routes:
    - when:
        - publish:
          - topic: taxi/{id}/location
            params:
              id: ${guarded['jwt'].identity}
        - subscribe:
          - topic: taxi/{id}/update
            params:
              id: ${guarded['jwt'].identity}
      exit: mqtt_kafka_proxy0

Fixes #1382

[jfallows]

We often test via spec scripts to verify implementation behavior instead of tightly coupled unit tests.
The intent is to be able to refactor the code, but still have stable tests to verify new implementation. Tightly coupled tests are typically forced to change when the code is refactored, preventing the stable basis to verify consistency.

Let's discuss further over Slack community as needed on this feedback.

[jfallows]

See HttpOptionsConfigAdapter for an example of a fairly complex object adapter so you can see the coding patterns.

[epieffe]

Ok, I updated the MqttConditionConfigAdapter class to align with the conding patterns found in other Adapter classes.

Let me know if you like it, thanks!

[jfallows]

Suggested change

	if (subscribeJson.containsKey(PARAMS_NAME))
	{
	subscribeJson.getJsonObject(PARAMS_NAME).forEach((n, v) -> subscribe.param()
	.name(n)
	.value(JsonString.class.cast(v).getString())
	.build());
	}
	if (subscribeJson.containsKey(PARAMS_NAME))
	{
	JsonObject paramsJson = subscribeJson.getJsonObject(PARAMS_NAME);
	paramsJson.keys().forEach(n ->
	subscribe.param()
	.name(n)
	.value(paramsJson.getString(n))
	.build());
	}

See https://www.javadoc.io/doc/jakarta.json/jakarta.json-api/1.1.6/javax/json/JsonObject.html#getString(java.lang.String)

[jfallows]

Same feedback as above for subscribe params.

[jfallows]

Suggested change

	private final List<LongObjPredicate<String>> subscribeMatchPredicates;
	private final List<LongObjPredicate<String>> publishMatchPredicates;
	private final List<TopicMatcher> subscribeMatchers;
	private final List<TopicMatcher> publishMatchers;

and create a private non-static inner class TopicMatcher.

private final class TopicMatcher
{
    private final Matcher matchTopic;
    private final Map<String, LongObjPredicate> matchParams;

    private TopicMatcher(
        MqttRouteConfig route,
        String wildcard,
        List<MqttParamConfig> params)
    {
        this.matchTopic = Pattern.compile(wildcard
                .replace(".", "\\.")
                .replace("$", "\\$")
                .replace("+", "[^/]*")
                .replace("#", ".*")
                .replaceAll("\\{([a-zA-Z_]+)\\}", "(?<$1>[^/]+)")).matcher("");
        this.matchParams = params != null
            ? params.stream()
                .collect(Collectors.toMap(p -> p.name, p -> asTopicParamMatcher(route.identities::get, p.value))
            : null;
    }

    private boolean matches(
        long authorization,
        String topic)
    {
        return matchTopic.reset(topic).matches() &&
            matchParams(matchTopic::group, authorization);
    }

    private boolean matchParams(
        UnaryFunction<String> valuesByName
        long authorization)
    {
        return matchParams == null ||
            matchParams.entrySet().stream()
                .allMatch(e -> e.getValue().test(authorization, valuesByName.apply(e.getKey())));
    }

    private LongObjPredicate asTopicParamMatcher(
        Function<String, LongFunction> identities,
        String value)
    {
        return (identityMatcher.reset(param.value).matches())
            ? asTopicParamIdentityMatcher(identities.apply(identityMatcher.group(1)))
            : asTopicParamValueMatcher(value);
    }

    private LongObjPredicate asTopicParamIdentityMatcher(
        LongFunction<String> identity)
    {
        return (a, v) -> v != null && identity != null && v.equals(identity.apply(a));
    }

    private LongObjPredicate asTopicParamMatcher(
        String expected)
    {
        return (a, v) -> v != null && v.equals(expected);
    }
}

Something more like this perhaps.

[jfallows]

Move this to a field on the class instead of passing to asTopicMatchPredicate(...).

private final Matcher identityMatcher = IDENTITY_PATTERN.matcher("");

[jfallows]

Suggested change

	this.subscribeMatchPredicates =
	condition.subscribes != null && !condition.subscribes.isEmpty() ?
	asTopicMatcher(condition.subscribes.stream().map(s -> s.topic).collect(Collectors.toList())) : null;
	this.publishMatchers =
	condition.subscribes.stream().map(s -> asTopicMatchPredicate(s.topic, s.params, route, identityMatcher))
	.collect(Collectors.toList()) : null;
	this.subscribeMatchers = condition.subscribes != null && !condition.subscribes.isEmpty()
	? condition.subscribes.stream().map(s -> new TopicMatcher(route::identity, s.topic, s.params)).toList()
	: null;

[jfallows]

Suggested change

	this.publishMatchPredicates =
	condition.publishes != null && !condition.publishes.isEmpty() ?
	asTopicMatcher(condition.publishes.stream().map(s -> s.topic).collect(Collectors.toList())) : null;
	condition.publishes.stream().map(s -> asTopicMatchPredicate(s.topic, s.params, route, identityMatcher))
	.collect(Collectors.toList()) : null;
	this.publishMatchers = condition.publishes != null && ! condition.publishes.isEmpty()
	? condition.publishes.stream().map(p -> new TopicMatcher(route::identity, p.topic, p.params)).toList()
	: null;

[jfallows]

Lookup by guard name can be moved to parse time as we know the guard name when parsing the value expression.

So this needs to return the LongFunction instead.

Suggested change

	String identity(
	String guard,
	long authorization)
	{
	return identities.getOrDefault(guard, a -> null).apply(authorization);
	}
	String identity(
	String guard)
	{
	return identities.get(guard);
	}

Alternatively, we could move identities to public scope and just use route.identities::get instead.