001/* 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.juneau.console.bus; 018 019import static org.apache.juneau.commons.utils.Shorts.*; 020 021import java.time.*; 022import java.util.*; 023import java.util.function.*; 024import java.util.regex.*; 025 026import org.apache.juneau.rest.server.filter.*; 027 028import jakarta.servlet.http.*; 029 030/** 031 * What a {@link ServerBus} will carry, in which direction, to whom. Everything not allowed here is denied. 032 * 033 * <p> 034 * Downstream entries are topic patterns: a custom topic ({@code ns.name}), a keyed custom topic or key pattern 035 * ({@code ns.name:key}, {@code ns.name:*}), a job topic or pattern ({@code job:<id>}, {@code job:*}), or a card command 036 * topic ({@code cmd:<cardId>}). Upstream entries are exact custom topics, each with its {@link UpstreamHandler}. 037 * 038 * <h5 class='section'>Example:</h5> 039 * <p class='bjava'> 040 * BusPolicy <jv>p</jv> = BusPolicy.<jsm>create</jsm>() 041 * .downstream(<js>"ops.jobs"</js>, <jk>true</jk>) <jc>// retained state</jc> 042 * .downstream(<js>"ops.alert:*"</js>, <jk>false</jk>) <jc>// events, any key</jc> 043 * .upstream(<js>"ops.cancel-all"</js>, (<jv>session</jv>, <jv>topic</jv>, <jv>payload</jv>) -> <jv>jobs</jv>.cancelAll()) 044 * .authorizer((<jv>req</jv>, <jv>topic</jv>) -> <jv>req</jv>.isUserInRole(<js>"ops"</js>)) <jc>// per-user, both directions</jc> 045 * .boundary(<jv>boundary</jv>) <jc>// LoopbackBoundary or CsrfBoundary</jc> 046 * .build(); 047 * 048 * <jc>// In a page test: every topic the page's bridges request is granted (E-55 otherwise).</jc> 049 * <jv>p</jv>.check(PageCapture.<jsm>of</jsm>(<jv>html</jv>).contract()); 050 * </p> 051 * 052 * @since 10.0.0 053 */ 054public final class BusPolicy { 055 056 static final String CUSTOM = "[a-z][a-z0-9-]{0,31}\\.[a-z][a-z0-9-]{0,31}"; 057 static final String KEY = "[A-Za-z0-9_.-]{1,128}"; 058 static final String CARD_ID = "[A-Za-z][A-Za-z0-9_-]{0,63}"; 059 060 /** Valid downstream topic patterns. */ 061 static final Pattern DOWNSTREAM_PATTERN = Pattern.compile( 062 "^(" + CUSTOM + "(:(" + KEY + "|\\*))?|job:(" + KEY + "|\\*)|cmd:" + CARD_ID + ")$"); 063 064 /** Valid upstream topics: exact custom topics only. */ 065 static final Pattern UPSTREAM_TOPIC = Pattern.compile("^" + CUSTOM + "(:" + KEY + ")?$"); 066 067 /** The C1 id pattern, used for bridge ids. */ 068 static final Pattern BRIDGE_ID = Pattern.compile("^" + CARD_ID + "$"); 069 070 private final Map<String,Boolean> downstream; 071 private final Map<String,UpstreamHandler> upstream; 072 private final BiPredicate<HttpServletRequest,String> authorizer; 073 private final RequestBoundary boundary; 074 private final List<String> allowedOrigins; 075 private final Duration sessionGrace; 076 private final Duration maxSessionAge; 077 private final Duration heartbeat; 078 private final int maxSessions; 079 private final int maxQueuedFrames; 080 private final int maxFrameBytes; 081 private final int maxUpstreamPerSecond; 082 private final Function<HttpServletRequest,Map<String,Object>> sessionAttributes; 083 084 private BusPolicy(Builder b) { 085 downstream = Collections.unmodifiableMap(new LinkedHashMap<>(b.downstream)); 086 upstream = Collections.unmodifiableMap(new LinkedHashMap<>(b.upstream)); 087 authorizer = b.authorizer; 088 boundary = b.boundary; 089 allowedOrigins = b.allowedOrigins; 090 sessionGrace = b.sessionGrace; 091 maxSessionAge = b.maxSessionAge; 092 heartbeat = b.heartbeat; 093 maxSessions = b.maxSessions; 094 maxQueuedFrames = b.maxQueuedFrames; 095 maxFrameBytes = b.maxFrameBytes; 096 maxUpstreamPerSecond = b.maxUpstreamPerSecond; 097 sessionAttributes = b.sessionAttributes; 098 } 099 100 /** 101 * Creates a new builder. An unmodified builder builds a policy that grants nothing. 102 * 103 * @return A new builder. 104 */ 105 public static Builder create() { 106 return new Builder(); 107 } 108 109 /** 110 * Returns whether a topic (or a requested pattern) may flow downstream, and if so whether it is retained. 111 * 112 * <p> 113 * A policy entry covers a topic when it is equal to it, or when the entry is {@code family:*} and the topic is 114 * {@code family:<anything>}. 115 * 116 * @param topic A concrete topic or a requested pattern. Can be <jk>null</jk>. 117 * @return The retain flag, or empty when the topic is denied. 118 */ 119 public Optional<Boolean> downstreamRetain(String topic) { 120 if (topic == null || ! DOWNSTREAM_PATTERN.matcher(topic).matches()) 121 return Optional.empty(); 122 var exact = downstream.get(topic); 123 if (exact != null) 124 return Optional.of(exact); 125 var i = topic.indexOf(':'); 126 if (i > 0) { 127 var wild = downstream.get(topic.substring(0, i) + ":*"); 128 if (wild != null) 129 return Optional.of(wild); 130 } 131 return Optional.empty(); 132 } 133 134 /** 135 * Returns whether a topic may flow upstream (exact match only). 136 * 137 * @param topic The topic. Can be <jk>null</jk>. 138 * @return <jk>true</jk> if the policy has a handler for it. 139 */ 140 public boolean allowsUpstream(String topic) { 141 return topic != null && upstream.containsKey(topic); 142 } 143 144 UpstreamHandler upstreamHandler(String topic) { 145 return upstream.get(topic); 146 } 147 148 /** 149 * Checks that this policy grants every topic that a page contract's {@code bridges} request (E-55). 150 * 151 * <p> 152 * Adopter page tests call this next to {@code PageContractAssert.hasNoWiringErrors()}. The request-dependent 153 * {@link #authorizer()} is not consulted: it can only narrow a grant at session time. 154 * 155 * @param contract The page contract (for example {@code PageCapture.of(html).contract()}). 156 * @throws IllegalStateException Listing every denied topic, one {@code E-55} line each. 157 */ 158 @SuppressWarnings({ 159 "java:S3776" // Walking the bridge contract is inherently nested. 160 }) 161 public void check(Map<String,?> contract) { 162 var problems = new ArrayList<String>(); 163 if (contract != null && contract.get("bridges") instanceof Collection<?> bridges) { 164 for (var raw : bridges) { 165 if (! (raw instanceof Map<?,?> bridge)) 166 continue; 167 var id = String.valueOf(bridge.get("id")); 168 for (var t : strings(bridge.get("downstream"))) 169 if (downstreamRetain(t).isEmpty()) 170 problems.add(e55("downstream", t, id)); 171 for (var t : strings(bridge.get("upstream"))) 172 if (! allowsUpstream(t)) 173 problems.add(e55("upstream", t, id)); 174 } 175 } 176 if (! problems.isEmpty()) 177 throw new IllegalStateException(String.join("\n", problems)); 178 } 179 180 private static String e55(String direction, String topic, String bridge) { 181 return "E-55: " + f("BusPolicy denies %s topic '%s' that bridge '%s' requests", direction, topic, bridge); 182 } 183 184 private static List<String> strings(Object o) { 185 return o instanceof Collection<?> c ? c.stream().map(String::valueOf).toList() : List.of(); 186 } 187 188 /** @return The per-request, per-topic authorizer applied at session open. Defaults to allow-all. */ 189 public BiPredicate<HttpServletRequest,String> authorizer() { return authorizer; } 190 191 /** @return The request boundary, or <jk>null</jk> when the application has its own guard. */ 192 public RequestBoundary boundary() { return boundary; } 193 194 /** @return The exact origins allowed on a WebSocket handshake when no {@link #boundary()} is set. */ 195 public List<String> allowedOrigins() { return allowedOrigins; } 196 197 /** @return How long a session survives after its last connection closes. Default 60 s. */ 198 public Duration sessionGrace() { return sessionGrace; } 199 200 /** @return The absolute session lifetime. Default 12 h. */ 201 public Duration maxSessionAge() { return maxSessionAge; } 202 203 /** @return The idle interval after which a {@code ping} frame is sent. Default 15 s. */ 204 public Duration heartbeat() { return heartbeat; } 205 206 /** @return The maximum number of live sessions. Default 256. */ 207 public int maxSessions() { return maxSessions; } 208 209 /** @return The per-connection outbound queue bound. Default 1024. */ 210 public int maxQueuedFrames() { return maxQueuedFrames; } 211 212 /** @return The maximum encoded frame size in UTF-8 bytes, both directions. Default 65536. */ 213 public int maxFrameBytes() { return maxFrameBytes; } 214 215 /** @return The per-session upstream frame rate limit. Default 20. */ 216 public int maxUpstreamPerSecond() { return maxUpstreamPerSecond; } 217 218 /** @return The function that captures session attributes at session open. Defaults to an empty map. */ 219 public Function<HttpServletRequest,Map<String,Object>> sessionAttributes() { return sessionAttributes; } 220 221 /** 222 * Builder for {@link BusPolicy}. 223 * 224 * <h5 class='section'>Example:</h5> 225 * <p class='bjava'> 226 * BusPolicy <jv>p</jv> = BusPolicy.<jsm>create</jsm>() 227 * .downstream(<js>"ops.jobs"</js>, <jk>true</jk>) 228 * .sessionAttributes(<jv>req</jv> -> Map.<jsm>of</jsm>(<js>"region"</js>, <jv>req</jv>.getHeader(<js>"X-Region"</js>))) 229 * .maxSessions(64) 230 * .build(); 231 * </p> 232 */ 233 public static final class Builder { 234 235 final Map<String,Boolean> downstream = new LinkedHashMap<>(); 236 final Map<String,UpstreamHandler> upstream = new LinkedHashMap<>(); 237 BiPredicate<HttpServletRequest,String> authorizer = (req, topic) -> true; 238 RequestBoundary boundary; 239 List<String> allowedOrigins = List.of(); 240 Duration sessionGrace = Duration.ofSeconds(60); 241 Duration maxSessionAge = Duration.ofHours(12); 242 Duration heartbeat = Duration.ofSeconds(15); 243 int maxSessions = 256; 244 int maxQueuedFrames = 1024; 245 int maxFrameBytes = 65536; 246 int maxUpstreamPerSecond = 20; 247 Function<HttpServletRequest,Map<String,Object>> sessionAttributes = req -> Map.of(); 248 249 Builder() {} 250 251 /** 252 * Allows a topic pattern downstream (server → page). 253 * 254 * @param topicPattern {@code ns.name}, {@code ns.name:key}, {@code ns.name:*}, {@code job:<id>}, {@code job:*} or {@code cmd:<cardId>}. 255 * @param retain Whether values are retained and resent on every (re)connect. 256 * @return This object. 257 * @throws IllegalArgumentException E-57 on an invalid or repeated pattern. 258 */ 259 public Builder downstream(String topicPattern, boolean retain) { 260 if (topicPattern == null || ! DOWNSTREAM_PATTERN.matcher(topicPattern).matches()) 261 throw e57("BusPolicy: '%s' is not a valid topic pattern", topicPattern); 262 if (downstream.putIfAbsent(topicPattern, retain) != null) 263 throw e57("BusPolicy: '%s' is declared twice", topicPattern); 264 return this; 265 } 266 267 /** 268 * Allows an exact custom topic upstream (page → server, WebSocket only) and names its handler. 269 * 270 * @param topic The topic ({@code ns.name} or {@code ns.name:key}). 271 * @param handler The handler. Must not be <jk>null</jk>. 272 * @return This object. 273 * @throws IllegalArgumentException E-57 on an invalid or repeated topic, or a <jk>null</jk> handler. 274 */ 275 public Builder upstream(String topic, UpstreamHandler handler) { 276 if (topic == null || ! UPSTREAM_TOPIC.matcher(topic).matches()) 277 throw e57("BusPolicy: '%s' is not a valid topic pattern", topic); 278 if (handler == null) 279 throw e57("BusPolicy: upstream topic '%s' has no handler", topic); 280 if (upstream.putIfAbsent(topic, handler) != null) 281 throw e57("BusPolicy: '%s' is declared twice", topic); 282 return this; 283 } 284 285 /** 286 * Sets a per-request, per-topic authorizer, consulted for every requested topic at session open. 287 * 288 * @param value The authorizer. Must not be <jk>null</jk>. 289 * @return This object. 290 */ 291 public Builder authorizer(BiPredicate<HttpServletRequest,String> value) { 292 authorizer = Objects.requireNonNull(value, "authorizer"); 293 return this; 294 } 295 296 /** 297 * Sets the request boundary. The session POST and SSE stream apply its {@code check}. 298 * 299 * <p> 300 * The WebSocket handshake cannot run {@code check}, so it applies the boundary's own handshake rules: a 301 * {@link LoopbackBoundary}'s exact {@code origin()} and {@code authority()}, or a {@link CsrfBoundary}'s 302 * {@code Origin} match against the handshake's own {@code Host} and scheme. {@code BusWebSockets.register} 303 * refuses any other boundary type. 304 * 305 * @param value The boundary. Can be <jk>null</jk>. 306 * @return This object. 307 */ 308 public Builder boundary(RequestBoundary value) { 309 boundary = value; 310 return this; 311 } 312 313 /** 314 * Sets the exact origins a WebSocket handshake may carry when no boundary is set. Ignored when a boundary is set. 315 * 316 * @param values Origins such as {@code https://console.example.org}. None may be blank. 317 * @return This object. 318 */ 319 public Builder allowedOrigins(String...values) { 320 for (var v : Objects.requireNonNull(values, "allowedOrigins")) 321 if (ib(v)) 322 throw new IllegalArgumentException("BusPolicy: allowedOrigins entries must not be blank"); 323 allowedOrigins = List.of(values); 324 return this; 325 } 326 327 /** @param value Grace after the last connection closes. Default 60 s. @return This object. */ 328 public Builder sessionGrace(Duration value) { sessionGrace = positive("sessionGrace", value); return this; } 329 330 /** @param value Absolute session lifetime. Default 12 h. @return This object. */ 331 public Builder maxSessionAge(Duration value) { maxSessionAge = positive("maxSessionAge", value); return this; } 332 333 /** @param value Idle interval before a {@code ping} frame. Default 15 s. @return This object. */ 334 public Builder heartbeat(Duration value) { heartbeat = positive("heartbeat", value); return this; } 335 336 /** @param value Maximum live sessions. Default 256. @return This object. */ 337 public Builder maxSessions(int value) { maxSessions = positive("maxSessions", value); return this; } 338 339 /** @param value Per-connection outbound queue bound. Default 1024. @return This object. */ 340 public Builder maxQueuedFrames(int value) { maxQueuedFrames = positive("maxQueuedFrames", value); return this; } 341 342 /** @param value Maximum encoded frame size, UTF-8 bytes. Default 65536. @return This object. */ 343 public Builder maxFrameBytes(int value) { maxFrameBytes = positive("maxFrameBytes", value); return this; } 344 345 /** @param value Per-session upstream frames per second. Default 20. @return This object. */ 346 public Builder maxUpstreamPerSecond(int value) { maxUpstreamPerSecond = positive("maxUpstreamPerSecond", value); return this; } 347 348 /** 349 * Sets the function that captures attributes onto each {@link BusSession} at session open. 350 * 351 * @param value The function. Must not be <jk>null</jk>; it may return <jk>null</jk> for "none". 352 * @return This object. 353 */ 354 public Builder sessionAttributes(Function<HttpServletRequest,Map<String,Object>> value) { 355 sessionAttributes = Objects.requireNonNull(value, "sessionAttributes"); 356 return this; 357 } 358 359 /** 360 * Builds the policy. 361 * 362 * @return A new immutable policy. 363 */ 364 public BusPolicy build() { 365 return new BusPolicy(this); 366 } 367 368 private static Duration positive(String name, Duration v) { 369 if (v == null || v.isZero() || v.isNegative()) 370 throw new IllegalArgumentException(f("BusPolicy: %s must be positive", name)); 371 return v; 372 } 373 374 private static int positive(String name, int v) { 375 if (v <= 0) 376 throw new IllegalArgumentException(f("BusPolicy: %s must be positive", name)); 377 return v; 378 } 379 380 private static IllegalArgumentException e57(String format, Object...args) { 381 return new IllegalArgumentException("E-57: " + f(format, args)); 382 } 383 } 384}