WebSocketRequestDecoder extends MessageToMessageDecoder<WebSocketFrame> { @Override protected void decode(ChannelHandlerContext ctx, WebSocketFrame msg, List<Object> out) throws Exception { if (msg instanceof TextWebSocketFrame) { TextWebSocketFrame frame = (TextWebSocketFrame) msg; String content = frame.text(); WebSocketRequest request = JsonUtil.getObjectMapper().readValue(content, WebSocketRequest.class); LOG.trace("Received WS request {}", content); String sessionId = ctx.channel().attr(SubscriptionRegistry.SESSION_ID_ATTR).get(); Multimap<String, String> headers = ctx.channel().attr(HTTP_HEADERS_ATTR).get(); X509Certificate clientCert = ctx.channel().attr(CLIENT_CERT_ATTR).get(); if (request instanceof AuthenticatedRequest) { if (headers != null) { ((AuthenticatedRequest) request).addHeaders(headers); } TimelyAuthenticationToken token; String authHeader = HttpHeaderUtils.getSingleHeader(headers, AUTH_HEADER, true); if (authHeader != null) { token = AuthenticationService.getAuthenticationToken(headers); } else if (StringUtils.isNotBlank(sessionId)) { TimelyPrincipal principal; ((AuthenticatedRequest) request).setSessionId(sessionId); if (AuthCache.containsKey(sessionId)) { principal = AuthCache.get(sessionId); } else { principal = TimelyPrincipal.anonymousPrincipal(); } SubjectIssuerDNPair dn = principal.getPrimaryUser().getDn(); token = AuthenticationService.getAuthenticationToken(null, dn.subjectDN(), dn.issuerDN()); } else if (clientCert != null) { token = AuthenticationService.getAuthenticationToken(clientCert, headers); } else { SubjectIssuerDNPair dn = TimelyPrincipal.anonymousPrincipal().getPrimaryUser().getDn(); token = AuthenticationService.getAuthenticationToken(null, dn.subjectDN(), dn.issuerDN()); } ((AuthenticatedRequest) request).setToken(token); } try { request.validate(); } catch (IllegalArgumentException e) { LOG.error("Error validating web socket request: " + e.getMessage()); ctx.writeAndFlush(new CloseWebSocketFrame(1008, e.getMessage())); return; } try { if (request instanceof AuthenticatedRequest) { try { AuthenticationService.enforceAccess((AuthenticatedRequest) request); } catch (TimelyException e) { if (!security.isAllowAnonymousWsAccess()) { throw e; } } } } catch (Exception e) { out.clear(); LOG.error("Error during access enforcment: " + e.getMessage()); ctx.writeAndFlush(new CloseWebSocketFrame(1008, e.getMessage())); return; } out.add(request); } else { LOG.error("Unhandled web socket frame type"); ctx.writeAndFlush(new CloseWebSocketFrame(1003, "Unhandled web socket frame type, only TextWebSocketFrame is supported")); } } WebSocketRequestDecoder(Security security); static void close(); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause); @Override void userEventTriggered(ChannelHandlerContext ctx, Object evt); static final AttributeKey<Multimap<String, String>> HTTP_HEADERS_ATTR; static final AttributeKey<X509Certificate> CLIENT_CERT_ATTR; }
@Test public void testVersion() throws Exception { decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); String request = "{ "+ "\"operation\" : \"version\"" + " }"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(VersionRequest.class, results.get(0).getClass()); }
@Test public void testPutMetric() throws Exception { decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); String request = "{" + "\"operation\" : \"put\",\n" + "\"name\" : \"sys.cpu.user\",\n" + "\"timestamp\":" + TEST_TIME + ",\n" + "\"measure\":1.0,\n" + "\"tags\":[ {\n" + "\"key\":\"tag1\",\n" + "\"value\":\"value1\"\n" + "},{\n" + "\"key\":\"tag2\",\n" + "\"value\":\"value2\"\n" + "}]\n" + "}"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(MetricRequest.class, results.get(0).getClass()); Metric metric = ((MetricRequest) results.iterator().next()).getMetric(); Assert.assertEquals("sys.cpu.user", metric.getName()); Assert.assertEquals(TEST_TIME, metric.getValue().getTimestamp()); Assert.assertEquals(1.0D, metric.getValue().getMeasure(), 0.0D); Assert.assertEquals(2, metric.getTags().size()); }
@Test public void testCreateSubscriptionWithMissingSessionId() throws Exception { decoder = new WebSocketRequestDecoder(config.getSecurity()); String request = "{ "+ "\"operation\" : \"create\", " + "\"subscriptionId\" : \"1234\"" + " }"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertNotNull(ctx.msg); Assert.assertEquals(CloseWebSocketFrame.class, ctx.msg.getClass()); Assert.assertEquals(1008, ((CloseWebSocketFrame) ctx.msg).statusCode()); Assert.assertEquals("User must authenticate", ((CloseWebSocketFrame) ctx.msg).reasonText()); }
@Test public void testCreateSubscriptionWithInvalidSessionIdAndNonAnonymousAccess() throws Exception { ctx.channel().attr(SubscriptionRegistry.SESSION_ID_ATTR) .set(URLEncoder.encode(UUID.randomUUID().toString(), StandardCharsets.UTF_8.name())); decoder = new WebSocketRequestDecoder(config.getSecurity()); String request = "{ "+ "\"operation\" : \"create\", " + "\"subscriptionId\" : \"1234\"" + " }"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertNotNull(ctx.msg); Assert.assertEquals(CloseWebSocketFrame.class, ctx.msg.getClass()); Assert.assertEquals(1008, ((CloseWebSocketFrame) ctx.msg).statusCode()); Assert.assertEquals("User must authenticate", ((CloseWebSocketFrame) ctx.msg).reasonText()); }
@Test public void testCreateSubscriptionWithValidSessionIdAndNonAnonymousAccess() throws Exception { ctx.channel().attr(SubscriptionRegistry.SESSION_ID_ATTR).set(cookie); decoder = new WebSocketRequestDecoder(config.getSecurity()); String request = "{ " + "\"operation\" : \"create\"," + "\"subscriptionId\" : \"" + cookie + "\"" + "}"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(CreateSubscription.class, results.get(0).getClass()); }
@Test public void testCreateSubscriptionWithoutSubscriptionId() throws Exception { decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); String request = "{ \"operation\" : \"create\" }"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertNotNull(ctx.msg); Assert.assertEquals(CloseWebSocketFrame.class, ctx.msg.getClass()); Assert.assertEquals(1008, ((CloseWebSocketFrame) ctx.msg).statusCode()); Assert.assertEquals("Subscription ID is required.", ((CloseWebSocketFrame) ctx.msg).reasonText()); }
@Test public void testCreateSubscription() throws Exception { decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); String request = "{ \"operation\" : \"create\", \"subscriptionId\" : \"1234\" }"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(CreateSubscription.class, results.get(0).getClass()); CreateSubscription create = (CreateSubscription) results.iterator().next(); create.validate(); }
@Test public void testAddSubscription() throws Exception { decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); String request = "{ \"operation\" : \"add\", \"subscriptionId\" : \"1234\" }"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(AddSubscription.class, results.get(0).getClass()); AddSubscription add = (AddSubscription) results.iterator().next(); add.validate(); }
@Test public void testRemoveSubscription() throws Exception { decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); String request = "{ \"operation\" : \"remove\", \"subscriptionId\" : \"1234\" }"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(RemoveSubscription.class, results.get(0).getClass()); RemoveSubscription remove = (RemoveSubscription) results.iterator().next(); remove.validate(); }
@Test public void testCloseSubscription() throws Exception { decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); String request = "{ \"operation\" : \"close\", \"subscriptionId\" : \"1234\" }"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(CloseSubscription.class, results.get(0).getClass()); CloseSubscription close = (CloseSubscription) results.iterator().next(); close.validate(); }
@Test public void testAggregrators() throws Exception { decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); String request = "{ \"operation\" : \"aggregators\", \"sessionId\" : \"1234\" }"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(AggregatorsRequest.class, results.get(0).getClass()); AggregatorsRequest agg = (AggregatorsRequest) results.iterator().next(); agg.validate(); }
@Test public void testMetrics() throws Exception { decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); String request = "{ \"operation\" : \"metrics\", \"sessionId\" : \"1234\" }"; TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(MetricsRequest.class, results.get(0).getClass()); MetricsRequest metrics = (MetricsRequest) results.iterator().next(); metrics.validate(); }
@Test public void testQuery() throws Exception { String request = "{\n"+ " \"operation\" : \"query\",\n"+ " \"sessionId\" : \"1234\",\n"+ " \"start\": 1356998400,\n"+ " \"end\": 1356998460,\n"+ " \"queries\": [\n"+ " {\n"+ " \"aggregator\": \"sum\",\n"+ " \"metric\": \"sys.cpu.user\",\n"+ " \"rate\": \"true\",\n"+ " \"rateOptions\": \n"+ " {\"counter\":false,\"counterMax\":100,\"resetValue\":0},\n"+ " \"tags\": {\n"+ " \"host\": \"*\",\n" + " \"rack\": \"r1\"\n" + " },\n"+ " \"filters\": [\n"+ " {\n"+ " \"type\":\"wildcard\",\n"+ " \"tagk\":\"host\",\n"+ " \"filter\":\"*\",\n"+ " \"groupBy\":true\n"+ " },\n"+ " {\n"+ " \"type\":\"literal_or\",\n"+ " \"tagk\":\"rack\",\n"+ " \"filter\":\"r1|r2\",\n"+ " \"groupBy\":false\n"+ " }\n"+ " ]\n"+ " },\n"+ " {\n"+ " \"aggregator\": \"sum\",\n"+ " \"tsuids\": [\n"+ " \"000001000002000042\",\n"+ " \"000001000002000043\"\n"+ " ]\n"+ " }\n"+ " ]\n"+ "}"; decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(QueryRequest.class, results.get(0).getClass()); QueryRequest query = (QueryRequest) results.iterator().next(); query.validate(); }
@Test public void testLookup() throws Exception { String request = "{ \"operation\" : \"lookup\", \"sessionId\" : \"1234\", \"metric\" : \"sys.cpu.user\" }"; decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(SearchLookupRequest.class, results.get(0).getClass()); SearchLookupRequest lookup = (SearchLookupRequest) results.iterator().next(); Assert.assertEquals("sys.cpu.user", lookup.getQuery()); lookup.validate(); }
@Test public void testSuggest() throws Exception { String request = "{\n" + " \"operation\" : \"suggest\",\n" + " \"sessionId\" : \"1234\",\n" + " \"type\": \"metrics\",\n" + " \"m\": \"sys.cpu.user\",\n" + " \"max\": 30\n" + "}"; decoder = new WebSocketRequestDecoder(anonConfig.getSecurity()); TextWebSocketFrame frame = new TextWebSocketFrame(); frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8)); decoder.decode(ctx, frame, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(SuggestRequest.class, results.get(0).getClass()); SuggestRequest suggest = (SuggestRequest) results.iterator().next(); Assert.assertEquals("metrics", suggest.getType()); Assert.assertEquals("sys.cpu.user", suggest.getMetric().get()); Assert.assertEquals(30, suggest.getMax()); suggest.validate(); }