From c49000f44b2216d15c9ad9370d3efb24f9a492cb Mon Sep 17 00:00:00 2001 From: Gopal Mahajan Date: Sat, 3 Sep 2022 03:36:00 +0530 Subject: [PATCH] immudb-read-flow-refactored --- .../rs/proxy/apiserver/ApiServerVerticle.java | 109 ++++++- .../proxy/apiserver/handlers/AuthHandler.java | 12 +- .../apiserver/util/ApiServerConstants.java | 64 +--- .../rs/proxy/authenticator/Constants.java | 3 +- .../JwtAuthenticationServiceImpl.java | 304 +++++++----------- .../authenticator/authorization/Api.java | 7 +- .../AuthorizationContextFactory.java | 4 + .../authorization/ConsumerAuthStrategy.java | 3 +- .../authenticator/authorization/IudxRole.java | 3 +- .../authorization/ProviderAuthStrategy.java | 24 ++ .../iudx/rs/proxy/common/ResponseUrn.java | 5 +- .../proxy/metering/MeteringServiceImpl.java | 222 ++++++++----- .../rs/proxy/metering/util/Constants.java | 56 ++-- .../rs/proxy/metering/util/QueryBuilder.java | 190 ++++++++--- .../proxy/metering/util/ResponseBuilder.java | 58 ---- .../metering/MeteringServiceImplTest.java | 116 +++++-- .../proxy/metering/util/QueryBuilderTest.java | 12 - .../metering/util/ResponseBuilderTest.java | 30 -- 18 files changed, 665 insertions(+), 557 deletions(-) create mode 100644 src/main/java/iudx/rs/proxy/authenticator/authorization/ProviderAuthStrategy.java delete mode 100644 src/main/java/iudx/rs/proxy/metering/util/ResponseBuilder.java delete mode 100644 src/test/java/iudx/rs/proxy/metering/util/QueryBuilderTest.java delete mode 100644 src/test/java/iudx/rs/proxy/metering/util/ResponseBuilderTest.java diff --git a/src/main/java/iudx/rs/proxy/apiserver/ApiServerVerticle.java b/src/main/java/iudx/rs/proxy/apiserver/ApiServerVerticle.java index 69061ec..55b7892 100644 --- a/src/main/java/iudx/rs/proxy/apiserver/ApiServerVerticle.java +++ b/src/main/java/iudx/rs/proxy/apiserver/ApiServerVerticle.java @@ -10,6 +10,8 @@ import static iudx.rs.proxy.apiserver.util.ApiServerConstants.HEADER_HOST; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.ID; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.IUDXQUERY_OPTIONS; +import static iudx.rs.proxy.apiserver.util.ApiServerConstants.IUDX_CONSUMER_AUDIT_URL; +import static iudx.rs.proxy.apiserver.util.ApiServerConstants.IUDX_PROVIDER_AUDIT_URL; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.JSON_COUNT; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.JSON_INSTANCEID; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.JSON_TYPE; @@ -23,6 +25,8 @@ import static iudx.rs.proxy.common.ResponseUrn.BACKING_SERVICE_FORMAT_URN; import static iudx.rs.proxy.common.ResponseUrn.INVALID_PARAM_URN; import static iudx.rs.proxy.common.ResponseUrn.INVALID_TEMPORAL_PARAM_URN; +import static iudx.rs.proxy.metering.util.Constants.RESULTS; +import static iudx.rs.proxy.metering.util.Constants.TOTAL_HITS; import io.netty.handler.codec.http.HttpConstants; import io.netty.handler.codec.http.QueryStringDecoder; @@ -57,6 +61,7 @@ import iudx.rs.proxy.metering.MeteringService; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -106,8 +111,7 @@ public void start() throws Exception { keystore = config().getString("keystore"); keystorePassword = config().getString("keystorePassword"); - serverOptions - .setSsl(true) + serverOptions.setSsl(true) .setKeyStoreOptions(new JksOptions().setPath(keystore).setPassword(keystorePassword)); } else { @@ -128,22 +132,21 @@ public void start() throws Exception { FailureHandler validationsFailureHandler = new FailureHandler(); ValidationHandler entityValidationHandler = new ValidationHandler(vertx, RequestType.ENTITY); - router - .get(NGSILD_ENTITIES_URL) - .handler(entityValidationHandler) - .handler(AuthHandler.create(vertx)) - .handler(this::handleEntitiesQuery) + router.get(NGSILD_ENTITIES_URL).handler(entityValidationHandler) + .handler(AuthHandler.create(vertx)).handler(this::handleEntitiesQuery) .failureHandler(validationsFailureHandler); ValidationHandler temporalValidationHandler = new ValidationHandler(vertx, RequestType.TEMPORAL); - router - .get(NGSILD_TEMPORAL_URL) - .handler(temporalValidationHandler) - .handler(AuthHandler.create(vertx)) - .handler(this::handleTemporalQuery) + router.get(NGSILD_TEMPORAL_URL).handler(temporalValidationHandler) + .handler(AuthHandler.create(vertx)).handler(this::handleTemporalQuery) .failureHandler(validationsFailureHandler); + + router.get(IUDX_CONSUMER_AUDIT_URL).handler(AuthHandler.create(vertx)) + .handler(this::getConsumerAuditDetail); + router.get(IUDX_PROVIDER_AUDIT_URL).handler(AuthHandler.create(vertx)) + .handler(this::getProviderAuditDetail); } private void handleEntitiesQuery(RoutingContext routingContext) { @@ -359,6 +362,88 @@ private void processBackendResponse(HttpServerResponse response, String failureM } } + private Future getConsumerAuditDetail(RoutingContext routingContext) { + LOGGER.debug("Info: getConsumerAuditDetail Started. "); + Promise promise = Promise.promise(); + JsonObject entries = new JsonObject(); + JsonObject consumer = (JsonObject) routingContext.data().get("authInfo"); + HttpServerRequest request = routingContext.request(); + HttpServerResponse response = routingContext.response(); + + entries.put("userid", consumer.getString("userid")); + entries.put("endPoint", consumer.getString("apiEndpoint")); + entries.put("startTime", request.getParam("time")); + entries.put("endTime", request.getParam("endTime")); + entries.put("timeRelation", request.getParam("timerel")); + entries.put("options", request.headers().get("options")); + entries.put("resourceId", request.getParam("id")); + entries.put("api", request.getParam("api")); + + { + LOGGER.debug(entries); + meteringService.executeReadQuery(entries, handler -> { + if (handler.succeeded()) { + LOGGER.info("Success: Search Success "); + if (Objects.equals(request.headers().get("options"), "count") && Integer.parseInt( + handler.result().getJsonArray(RESULTS).getJsonObject(0).getString(TOTAL_HITS)) == 0) { + handleSuccessResponse(response, ResponseType.NoContent.getCode(), + handler.result().toString()); + } else { + LOGGER.debug("Table Reading Done."); + handleSuccessResponse(response, ResponseType.Ok.getCode(), handler.result().toString()); + } + } else if (handler.failed()) { + LOGGER.error("Fail: Search Fail " + handler.cause().getMessage()); + LOGGER.debug(handler instanceof ServiceException); + processBackendResponse(response, handler.cause().getMessage()); + } + }); + return promise.future(); + } + } + + private Future getProviderAuditDetail(RoutingContext routingContext) { + LOGGER.trace("Info: getProviderAuditDetail Started."); + Promise promise = Promise.promise(); + JsonObject entries = new JsonObject(); + JsonObject provider = (JsonObject) routingContext.data().get("authInfo"); + HttpServerRequest request = routingContext.request(); + HttpServerResponse response = routingContext.response(); + + entries.put("endPoint", provider.getString("apiEndpoint")); + entries.put("userid", provider.getString("userid")); + entries.put("iid", provider.getString("iid")); + entries.put("startTime", request.getParam("time")); + entries.put("endTime", request.getParam("endTime")); + entries.put("timeRelation", request.getParam("timerel")); + entries.put("providerID", request.getParam("providerID")); + entries.put("consumerID", request.getParam("consumer")); + entries.put("resourceId", request.getParam("id")); + entries.put("api", request.getParam("api")); + entries.put("options", request.headers().get("options")); + + { + LOGGER.debug(entries); + meteringService.executeReadQuery(entries, handler -> { + if (handler.succeeded()) { + LOGGER.info("Success: Search Success "); + if (Objects.equals(request.headers().get("options"), "count") && Integer.parseInt( + handler.result().getJsonArray(RESULTS).getJsonObject(0).getString(TOTAL_HITS)) == 0) { + handleSuccessResponse(response, ResponseType.NoContent.getCode(), + handler.result().toString()); + } else { + handleSuccessResponse(response, ResponseType.Ok.getCode(), handler.result().toString()); + } + } else if (handler.failed()) { + LOGGER.error("Fail: Search Fail " + handler.cause().getMessage()); + LOGGER.debug(handler instanceof ServiceException); + processBackendResponse(response, handler.cause().getMessage()); + } + }); + return promise.future(); + } + } + private void updateAuditTable(RoutingContext context) { Promise promise = Promise.promise(); JsonObject authInfo = (JsonObject) context.data().get("authInfo"); diff --git a/src/main/java/iudx/rs/proxy/apiserver/handlers/AuthHandler.java b/src/main/java/iudx/rs/proxy/apiserver/handlers/AuthHandler.java index 3eb7e74..2c85981 100644 --- a/src/main/java/iudx/rs/proxy/apiserver/handlers/AuthHandler.java +++ b/src/main/java/iudx/rs/proxy/apiserver/handlers/AuthHandler.java @@ -5,11 +5,12 @@ import static iudx.rs.proxy.apiserver.util.ApiServerConstants.APPLICATION_JSON; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.CONTENT_TYPE; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.ENTITIES_URL_REGEX; -import static iudx.rs.proxy.apiserver.util.ApiServerConstants.EXPIRY; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.HEADER_TOKEN; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.ID; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.IDS; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.IID; +import static iudx.rs.proxy.apiserver.util.ApiServerConstants.IUDX_CONSUMER_AUDIT_URL; +import static iudx.rs.proxy.apiserver.util.ApiServerConstants.IUDX_PROVIDER_AUDIT_URL; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.JSON_DETAIL; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.JSON_TITLE; import static iudx.rs.proxy.apiserver.util.ApiServerConstants.JSON_TYPE; @@ -38,7 +39,7 @@ public class AuthHandler implements Handler { private static final Logger LOGGER = LogManager.getLogger(AuthHandler.class); - static AuthenticationService authenticator; + static AuthenticationService authenticator; private final String AUTH_INFO = "authInfo"; private HttpServerRequest request; @@ -95,7 +96,6 @@ public void handle(RoutingContext context) { return; } context.next(); - return; }); } @@ -127,7 +127,7 @@ private JsonObject generateResponse(ResponseUrn urn, HttpStatusCode statusCode) /** * extract id from request (path/query or body ) * - * @param ctx current routing context + * @param ctx current routing context * @param forPath endpoint called for * @return id extraced fro path if present */ @@ -159,6 +159,10 @@ private String getNormalizedPath(String url) { path = NGSILD_TEMPORAL_URL; } else if (url.matches(ENTITIES_URL_REGEX)) { path = NGSILD_ENTITIES_URL; + } else if (url.matches(IUDX_CONSUMER_AUDIT_URL)) { + path = IUDX_CONSUMER_AUDIT_URL; + } else if (url.matches(IUDX_PROVIDER_AUDIT_URL)) { + path = IUDX_PROVIDER_AUDIT_URL; } return path; } diff --git a/src/main/java/iudx/rs/proxy/apiserver/util/ApiServerConstants.java b/src/main/java/iudx/rs/proxy/apiserver/util/ApiServerConstants.java index 97f8908..067c6cd 100644 --- a/src/main/java/iudx/rs/proxy/apiserver/util/ApiServerConstants.java +++ b/src/main/java/iudx/rs/proxy/apiserver/util/ApiServerConstants.java @@ -1,10 +1,10 @@ package iudx.rs.proxy.apiserver.util; +import io.vertx.core.http.HttpMethod; import java.util.Arrays; import java.util.HashSet; import java.util.List; import java.util.Set; -import io.vertx.core.http.HttpMethod; import java.util.regex.Pattern; public class ApiServerConstants { @@ -34,17 +34,15 @@ public class ApiServerConstants { public static final String NGSILD_ENTITIES_URL = NGSILD_BASE_PATH + "/entities"; public static final String ENTITIES_URL_REGEX = NGSILD_ENTITIES_URL + "(.*)"; public static final String NGSILD_TEMPORAL_URL = NGSILD_BASE_PATH + "/temporal/entities"; - + // path regex + public static final String TEMPORAL_URL_REGEX = NGSILD_TEMPORAL_URL + "(.*)"; + public static final String IUDX_CONSUMER_AUDIT_URL = NGSILD_BASE_PATH + "/consumer/audit"; // date-time format - + public static final String IUDX_PROVIDER_AUDIT_URL = NGSILD_BASE_PATH + "/provider/audit"; public static final String API_ENDPOINT = "apiEndpoint"; public static final String API_METHOD = "method"; public static final String ID = "id"; public static final String IDS = "ids"; - - // path regex - public static final String TEMPORAL_URL_REGEX = NGSILD_TEMPORAL_URL + "(.*)"; - // ngsi-ld/IUDX query paramaters public static final String NGSILDQUERY_ID = "id"; public static final String NGSILDQUERY_IDPATTERN = "idpattern"; @@ -61,10 +59,8 @@ public class ApiServerConstants { public static final String NGSILDQUERY_GEOPROPERTY = "geoproperty"; public static final String NGSILDQUERY_TIMEPROPERTY = "timeproperty"; public static final String NGSILDQUERY_MAXDISTANCE = "maxdistance"; - public static final String NGSILDQUERY_MINDISTANCE = "mindistance"; public static final String IUDXQUERY_OPTIONS = "options"; public static final String NGSILDQUERY_ENTITIES = "entities"; - public static final String NGSILDQUERY_GEOQ = "geoQ"; public static final String NGSILDQUERY_TEMPORALQ = "temporalQ"; public static final String NGSILDQUERY_TIME_PROPERTY = "timeProperty"; public static final String NGSILDQUERY_FROM = "offset"; @@ -73,77 +69,29 @@ public class ApiServerConstants { //json fields public static final String JSON_INSTANCEID = "instanceID"; - public static final String JSON_CONSUMER = "consumer"; - public static final String JSON_PROVIDER = "provider"; public static final String JSON_TYPE = "type"; - public static final String JSON_NAME = "name"; public static final String JSON_ENTITIES = "entities"; public static final String JSON_ID = "id"; public static final String JSON_ATTRIBUTE_FILTER = "attrs"; - public static final String JSON_NEAR = "near"; - public static final String JSON_LAT = "lat"; - public static final String JSON_LON = "lon"; - public static final String JSON_RADIUS = "radius"; - public static final String JSON_GEOMETRY = "geometry"; - public static final String JSON_COORDINATES = "coordinates"; - public static final String JSON_GEOREL = "georel"; - public static final String JSON_WITHIN = "within"; - public static final String JSON_MAXDISTANCE = "maxdistance"; - public static final String JSON_MINDISTANCE = "mindistance"; public static final String JSON_DURING = "during"; public static final String JSON_TIME = "time"; public static final String JSON_ENDTIME = "endtime"; public static final String JSON_TIMEREL = "timerel"; public static final String JSON_ATTR_QUERY = "attr-query"; - public static final String JSON_GEOPROPERTY = "geoproperty"; public static final String JSON_ATTRIBUTE = "attribute"; public static final String JSON_OPERATOR = "operator"; public static final String JSON_VALUE = "value"; public static final String JSON_TITLE = "title"; public static final String JSON_DETAIL = "detail"; - public static final String JSON_EXCHANGE_NAME = "exchangeName"; - public static final String JSON_QUEUE_NAME = "queueName"; - public static final String JSON_VHOST_NAME = "vHostName"; - public static final String JSON_VHOST = "vHost"; - public static final String JSON_VHOST_ID = "vhostId"; - public static final String DOMAIN = "domain"; - public static final String USERSHA = "userSha"; - public static final String JSON_ALIAS = "alias"; - public static final String JSON_STREAMING_TYPE = "streaming"; - public static final String JSON_EXCHANGE = "exchange"; - public static final String JSON_QUEUE = "queue"; - public static final String JSON_USERNAME = "username"; - public static final String JSON_APIKEY = "apiKey"; - public static final String JSON_STATUS = "status"; - public static final String JSON_STATUS_HEARTBEAT = "heartbeat"; - public static final String JSON_STATUS_SERVERISSUE = "Server Issue"; - public static final String JSON_STATUS_DATAISSUE = "Server Issue"; - public static final String JSON_STREAMING_NAME = "test-streaming-name"; - public static final String JSON_SUBS_ID = "subscriptionID"; public static final String JSON_COUNT = "Count"; - public static final String JSON_URL = "url"; - public static final String JSON_METHOD = "method"; - public static final String JSON_PASSWORD = "password"; - public static final String RESOURCE_SERVER = "resourceServer"; - public static final String RESOURCE_GROUP = "resourceGroup"; - public static final String RESOURCE_NAME = "resourceName"; - public static final String EXPIRY = "expiry"; public static final String IID = "iid"; public static final String API = "api"; public static final String USER_ID = "userid"; - - // Geometry - public static final String GEOM_POINT = "point"; - // searchtype public static final String JSON_SEARCH_TYPE = "searchType"; public static final String JSON_TEMPORAL_SEARCH = "temporalSearch_"; - public static final String JSON_GEO_SEARCH = "geoSearch_"; - public static final String JSON_RESPONSE_FILTER_SEARCH = "responseFilter_"; public static final String JSON_ATTRIBUTE_SEARCH = "attributeSearch_"; - public static final String JSON_LATEST_SEARCH = "latestSearch_"; - // request/response params public static final String CONTENT_TYPE = "content-type"; @@ -167,7 +115,7 @@ public class ApiServerConstants { public static final Pattern ID_REGEX = Pattern.compile( "^[a-zA-Z0-9.]{4,100}/{1}[a-zA-Z0-9.]{4,100}/{1}[a-zA-Z.]{4,100}/{1}[a-zA-Z-_.]{4,100}/{1}[a-zA-Z0-9-_.]{4,100}$"); - public static final String RESPONSE_SIZE="response_size"; + public static final String RESPONSE_SIZE = "response_size"; public static final double VALIDATION_ALLOWED_DIST = 1000.0; public static final int VALIDATION_PAGINATION_LIMIT_MAX = 5000; diff --git a/src/main/java/iudx/rs/proxy/authenticator/Constants.java b/src/main/java/iudx/rs/proxy/authenticator/Constants.java index 8f9b0c9..5cc4ab3 100644 --- a/src/main/java/iudx/rs/proxy/authenticator/Constants.java +++ b/src/main/java/iudx/rs/proxy/authenticator/Constants.java @@ -3,11 +3,10 @@ import java.util.List; public class Constants { - public static final List OPEN_ENDPOINTS = List.of("/ngsi-ld/v1/temporal/entities","/ngsi-ld/v1/entities"); + public static final List OPEN_ENDPOINTS = List.of("/ngsi-ld/v1/temporal/entities","/ngsi-ld/v1/entities","/ngsi-ld/v1/consumer/audit"); public static final long CACHE_TIMEOUT_AMOUNT = 30; public static final String CAT_RSG_PATH = "/iudx/cat/v1/search"; public static final String CAT_ITEM_PATH = "/iudx/cat/v1/item"; - public static final String CAT_RESOURCE_GROUP_PATH="/iudx/cat/v1/list/resourceGroup"; public static final String JSON_USERID = "userid"; public static final String JSON_IID = "iid"; public static final String JSON_EXPIRY = "expiry"; diff --git a/src/main/java/iudx/rs/proxy/authenticator/JwtAuthenticationServiceImpl.java b/src/main/java/iudx/rs/proxy/authenticator/JwtAuthenticationServiceImpl.java index 12189c9..269ff68 100644 --- a/src/main/java/iudx/rs/proxy/authenticator/JwtAuthenticationServiceImpl.java +++ b/src/main/java/iudx/rs/proxy/authenticator/JwtAuthenticationServiceImpl.java @@ -45,33 +45,23 @@ public class JwtAuthenticationServiceImpl implements AuthenticationService { private static final Logger LOGGER = LogManager.getLogger(JwtAuthenticationServiceImpl.class); final JWTAuth jwtAuth; - WebClient catWebClient; final String host; final int port; final String path; final String audience; final CacheService cache; - + WebClient catWebClient; // resourceGroupCache will contain ACL info about all resource group in a resource server - Cache resourceGroupCache = - CacheBuilder.newBuilder() - .maximumSize(1000) - .expireAfterAccess(Constants.CACHE_TIMEOUT_AMOUNT, TimeUnit.MINUTES) - .build(); + Cache resourceGroupCache = CacheBuilder.newBuilder().maximumSize(1000) + .expireAfterAccess(Constants.CACHE_TIMEOUT_AMOUNT, TimeUnit.MINUTES).build(); // resourceIdCache will contain info about resources available(& their ACL) in resource server. - Cache resourceIdCache = - CacheBuilder.newBuilder() - .maximumSize(1000) - .expireAfterAccess(Constants.CACHE_TIMEOUT_AMOUNT, TimeUnit.MINUTES) - .build(); + Cache resourceIdCache = CacheBuilder.newBuilder().maximumSize(1000) + .expireAfterAccess(Constants.CACHE_TIMEOUT_AMOUNT, TimeUnit.MINUTES).build(); - JwtAuthenticationServiceImpl( - Vertx vertx, - final JWTAuth jwtAuth, - final JsonObject config, - final CacheService cacheService) { + JwtAuthenticationServiceImpl(Vertx vertx, final JWTAuth jwtAuth, final JsonObject config, + final CacheService cacheService) { this.jwtAuth = jwtAuth; this.audience = config.getString("host"); this.host = config.getString("catServerHost"); @@ -84,65 +74,46 @@ public class JwtAuthenticationServiceImpl implements AuthenticationService { this.cache = cacheService; } @Override - public AuthenticationService tokenIntrospect( - JsonObject request, JsonObject authenticationInfo, Handler> handler) { - LOGGER.info("AUTH started."); + public AuthenticationService tokenIntrospect(JsonObject request, JsonObject authenticationInfo, + Handler> handler) { String id = authenticationInfo.getString("id"); String token = authenticationInfo.getString("token"); - Future jwtDecodeFuture = decodeJwt(token); - ResultContainer result = new ResultContainer(); - jwtDecodeFuture - .compose( - decodeHandler -> { - result.jwtData = decodeHandler; - return isValidAudienceValue(result.jwtData); - }) - .compose( - audienceHandler -> { - if (!result.jwtData.getIss().equals(result.jwtData.getSub())) { - return isRevokedClientToken(result.jwtData); - } else { - return Future.succeededFuture(true); - } - }) - .compose( - revokeTokenHandler -> { - if (!result.jwtData.getIss().equals(result.jwtData.getSub())) { - return isOpenResource(id); - } else { - return Future.succeededFuture("OPEN"); - } - }) - .compose( - openResourceHandler -> { - result.isOpen = openResourceHandler.equalsIgnoreCase("OPEN"); - if (result.jwtData.getIss().equals(result.jwtData.getSub())) { - JsonObject jsonResponse = new JsonObject(); - jsonResponse.put(JSON_USERID, result.jwtData.getSub()); - jsonResponse.put( - JSON_EXPIRY, - (LocalDateTime.ofInstant( - Instant.ofEpochSecond( - Long.parseLong(result.jwtData.getExp().toString())), - ZoneId.systemDefault())) - .toString()); - return Future.succeededFuture(jsonResponse); - } else { - return validateAccess(result.jwtData, result.isOpen, authenticationInfo); - } - }) - .onSuccess( - successHandler -> { - handler.handle(Future.succeededFuture(successHandler)); - }) - .onFailure( - failureHandler -> { - LOGGER.error("error : " + failureHandler.getMessage()); - handler.handle(Future.failedFuture(failureHandler.getMessage())); - }); + jwtDecodeFuture.compose(decodeHandler -> { + result.jwtData = decodeHandler; + return isValidAudienceValue(result.jwtData); + }).compose(audienceHandler -> { + if (!result.jwtData.getIss().equals(result.jwtData.getSub())) { + return isRevokedClientToken(result.jwtData); + } else { + return Future.succeededFuture(true); + } + }).compose(revokeTokenHandler -> { + if (!result.jwtData.getIss().equals(result.jwtData.getSub())) { + return isOpenResource(id); + } else { + return Future.succeededFuture("OPEN"); + } + }).compose(openResourceHandler -> { + result.isOpen = openResourceHandler.equalsIgnoreCase("OPEN"); + if (result.jwtData.getIss().equals(result.jwtData.getSub())) { + JsonObject jsonResponse = new JsonObject(); + jsonResponse.put(JSON_USERID, result.jwtData.getSub()); + jsonResponse.put(JSON_EXPIRY, (LocalDateTime.ofInstant( + Instant.ofEpochSecond(Long.parseLong(result.jwtData.getExp().toString())), + ZoneId.systemDefault())).toString()); + return Future.succeededFuture(jsonResponse); + } else { + return validateAccess(result.jwtData, result.isOpen, authenticationInfo); + } + }).onSuccess(successHandler -> { + handler.handle(Future.succeededFuture(successHandler)); + }).onFailure(failureHandler -> { + LOGGER.error("error : " + failureHandler.getMessage()); + handler.handle(Future.failedFuture(failureHandler.getMessage())); + }); return this; } @@ -150,25 +121,20 @@ Future decodeJwt(String jwtToken) { Promise promise = Promise.promise(); TokenCredentials creds = new TokenCredentials(jwtToken); - jwtAuth - .authenticate(creds) - .onSuccess( - user -> { - JwtData jwtData = new JwtData(user.principal()); - jwtData.setExp(user.get("exp")); - jwtData.setIat(user.get("iat")); - promise.complete(jwtData); - }) - .onFailure( - err -> { - LOGGER.error("failed to decode/validate jwt token : " + err.getMessage()); - promise.fail("failed"); - }); + jwtAuth.authenticate(creds).onSuccess(user -> { + JwtData jwtData = new JwtData(user.principal()); + jwtData.setExp(user.get("exp")); + jwtData.setIat(user.get("iat")); + promise.complete(jwtData); + }).onFailure(err -> { + LOGGER.error("failed to decode/validate jwt token : " + err.getMessage()); + promise.fail("failed"); + }); return promise.future(); } //private Future isOpenResource(String id) - Future isOpenResource(String id) { + Future isOpenResource(String id) { LOGGER.trace("isOpenResource() started"); Promise promise = Promise.promise(); @@ -183,26 +149,19 @@ Future isOpenResource(String id) { if (idComponents.length < 4) { promise.fail("Not Found " + id); } - String groupId = - (idComponents.length == 4) - ? id - : String.join("/", Arrays.copyOfRange(idComponents, 0, 4)); + String groupId = (idComponents.length == 4) ? id : + String.join("/", Arrays.copyOfRange(idComponents, 0, 4)); // 1. check group accessPolicy. // 2. check resource exist, if exist set accessPolicy to group accessPolicy. else fail Future groupACLFuture = getGroupAccessPolicy(groupId); - groupACLFuture - .compose( - groupACLResult -> { - String groupPolicy = groupACLResult; - return isResourceExist(id, groupPolicy); - }) - .onSuccess( - handler -> promise.complete(resourceIdCache.getIfPresent(id))) - .onFailure( - handler -> { - LOGGER.error("cat response failed for Id : (" + id + ")" + handler.getCause()); - promise.fail("Not Found " + id); - }); + groupACLFuture.compose(groupACLResult -> { + String groupPolicy = groupACLResult; + return isResourceExist(id, groupPolicy); + }).onSuccess(handler -> promise.complete(resourceIdCache.getIfPresent(id))) + .onFailure(handler -> { + LOGGER.error("cat response failed for Id : (" + id + ")" + handler.getCause()); + promise.fail("Not Found " + id); + }); } return promise.future(); } @@ -221,9 +180,9 @@ Future isRevokedClientToken(JwtData jwtData) { String timestamp = responseJson.getString("value"); LocalDateTime revokedAt = ZonedDateTime.parse(timestamp).toLocalDateTime(); - LocalDateTime jwtIssuedAt = (LocalDateTime.ofInstant( - Instant.ofEpochSecond(jwtData.getIat()), - ZoneId.systemDefault())); + LocalDateTime jwtIssuedAt = + (LocalDateTime.ofInstant(Instant.ofEpochSecond(jwtData.getIat()), + ZoneId.systemDefault())); if (jwtIssuedAt.isBefore(revokedAt)) { LOGGER.error("Privileges for client are revoked."); @@ -241,8 +200,8 @@ Future isRevokedClientToken(JwtData jwtData) { return promise.future(); } - public Future validateAccess( - JwtData jwtData, boolean openResource, JsonObject authInfo) { + public Future validateAccess(JwtData jwtData, boolean openResource, + JsonObject authInfo) { LOGGER.trace("validateAccess() started"); Promise promise = Promise.promise(); String jwtId = jwtData.getIid().split(":")[1]; @@ -263,19 +222,16 @@ public Future validateAccess( AuthorizationStrategy authStrategy = AuthorizationContextFactory.create(role); LOGGER.info("strategy : " + authStrategy.getClass().getSimpleName()); JwtAuthorization jwtAuthStrategy = new JwtAuthorization(authStrategy); - LOGGER.info("auth strategy "+jwtAuthStrategy); + LOGGER.info("auth strategy " + jwtAuthStrategy); LOGGER.info("endPoint : " + authInfo.getString("apiEndpoint")); if (jwtAuthStrategy.isAuthorized(authRequest, jwtData)) { LOGGER.info("User access is allowed."); JsonObject jsonResponse = new JsonObject(); jsonResponse.put(JSON_USERID, jwtData.getSub()); jsonResponse.put(JSON_IID, jwtId); - jsonResponse.put( - JSON_EXPIRY, - (LocalDateTime.ofInstant( - Instant.ofEpochSecond(Long.parseLong(jwtData.getExp().toString())), - ZoneId.systemDefault())) - .toString()); + jsonResponse.put(JSON_EXPIRY, (LocalDateTime.ofInstant( + Instant.ofEpochSecond(Long.parseLong(jwtData.getExp().toString())), + ZoneId.systemDefault())).toString()); promise.complete(jsonResponse); } else { LOGGER.error("failed - no access provided to endpoint"); @@ -306,33 +262,27 @@ private Future isResourceExist(String id, String groupACL) { promise.complete(true); } else { LOGGER.info("Info : Cache miss : call cat server"); - catWebClient - .get(port, host, path) - .addQueryParam("property", "[id]") - .addQueryParam("value", "[[" + id + "]]") - .addQueryParam("filter", "[id]") - .expect(ResponsePredicate.JSON) - .send( - responseHandler -> { - if (responseHandler.failed()) { - promise.fail("false"); - } - HttpResponse response = responseHandler.result(); - JsonObject responseBody = response.bodyAsJsonObject(); - if (response.statusCode() != HttpStatus.SC_OK) { - promise.fail("false"); - } else if (!responseBody.getString("type").equals("urn:dx:cat:Success")) { - promise.fail("Not Found"); - return; - } else if (responseBody.getInteger("totalHits") == 0) { - LOGGER.error("Info: Resource ID invalid : Catalogue item Not Found"); - promise.fail("Not Found"); - } else { - LOGGER.debug("is Exist response : " + responseBody); - resourceIdCache.put(id, groupACL); - promise.complete(true); - } - }); + catWebClient.get(port, host, path).addQueryParam("property", "[id]") + .addQueryParam("value", "[[" + id + "]]").addQueryParam("filter", "[id]") + .expect(ResponsePredicate.JSON).send(responseHandler -> { + if (responseHandler.failed()) { + promise.fail("false"); + } + HttpResponse response = responseHandler.result(); + JsonObject responseBody = response.bodyAsJsonObject(); + if (response.statusCode() != HttpStatus.SC_OK) { + promise.fail("false"); + } else if (!responseBody.getString("type").equals("urn:dx:cat:Success")) { + promise.fail("Not Found"); + } else if (responseBody.getInteger("totalHits") == 0) { + LOGGER.error("Info: Resource ID invalid : Catalogue item Not Found"); + promise.fail("Not Found"); + } else { + LOGGER.debug("is Exist response : " + responseBody); + resourceIdCache.put(id, groupACL); + promise.complete(true); + } + }); } return promise.future(); } @@ -346,45 +296,37 @@ private Future getGroupAccessPolicy(String groupId) { promise.complete(groupACL); } else { LOGGER.info("Info : cache miss"); - catWebClient - .get(port, host, path) - .addQueryParam("property", "[id]") - .addQueryParam("value", "[[" + groupId + "]]") - .addQueryParam("filter", "[accessPolicy]") - .expect(ResponsePredicate.JSON) - .send( - httpResponseAsyncResult -> { - if (httpResponseAsyncResult.failed()) { - LOGGER.error(httpResponseAsyncResult.cause()); - promise.fail("Resource not found"); - return; - } - HttpResponse response = httpResponseAsyncResult.result(); - if (response.statusCode() != HttpStatus.SC_OK) { - promise.fail("Resource not found"); - return; - } - JsonObject responseBody = response.bodyAsJsonObject(); - if (!responseBody.getString("type").equals("urn:dx:cat:Success")) { - promise.fail("Resource not found"); - return; - } - String resourceACL = "SECURE"; - try { - resourceACL = - responseBody - .getJsonArray("results") - .getJsonObject(0) - .getString("accessPolicy"); - resourceGroupCache.put(groupId, resourceACL); - LOGGER.debug("Info: Group ID valid : Catalogue item Found"); - promise.complete(resourceACL); - } catch (Exception ignored) { - LOGGER.error("Info: Group ID invalid : Empty response in results from Catalogue", - ignored); - promise.fail("Resource not found"); - } - }); + catWebClient.get(port, host, path).addQueryParam("property", "[id]") + .addQueryParam("value", "[[" + groupId + "]]").addQueryParam("filter", "[accessPolicy]") + .expect(ResponsePredicate.JSON).send(httpResponseAsyncResult -> { + if (httpResponseAsyncResult.failed()) { + LOGGER.error(httpResponseAsyncResult.cause()); + promise.fail("Resource not found"); + return; + } + HttpResponse response = httpResponseAsyncResult.result(); + if (response.statusCode() != HttpStatus.SC_OK) { + promise.fail("Resource not found"); + return; + } + JsonObject responseBody = response.bodyAsJsonObject(); + if (!responseBody.getString("type").equals("urn:dx:cat:Success")) { + promise.fail("Resource not found"); + return; + } + String resourceACL = "SECURE"; + try { + resourceACL = + responseBody.getJsonArray("results").getJsonObject(0).getString("accessPolicy"); + resourceGroupCache.put(groupId, resourceACL); + LOGGER.debug("Info: Group ID valid : Catalogue item Found"); + promise.complete(resourceACL); + } catch (Exception ignored) { + LOGGER.error("Info: Group ID invalid : Empty response in results from Catalogue", + ignored); + promise.fail("Resource not found"); + } + }); } return promise.future(); } diff --git a/src/main/java/iudx/rs/proxy/authenticator/authorization/Api.java b/src/main/java/iudx/rs/proxy/authenticator/authorization/Api.java index c82dbe5..98271a7 100644 --- a/src/main/java/iudx/rs/proxy/authenticator/authorization/Api.java +++ b/src/main/java/iudx/rs/proxy/authenticator/authorization/Api.java @@ -4,6 +4,9 @@ public enum Api { ENTITIES("/ngsi-ld/v1/entities"), + CONSUMER_AUDIT("/ngsi-ld/v1/consumer/audit"), + + PROVIDER_AUDIT("/ngsi-ld/v1/provider/audit"), TEMPORAL("/ngsi-ld/v1/temporal/entities"); private final String endpoint; @@ -13,9 +16,7 @@ public enum Api { } public static Api fromEndpoint(final String endpoint) { - return Stream.of(values()) - .filter(v -> v.endpoint.equalsIgnoreCase(endpoint)) - .findAny() + return Stream.of(values()).filter(v -> v.endpoint.equalsIgnoreCase(endpoint)).findAny() .orElse(null); } diff --git a/src/main/java/iudx/rs/proxy/authenticator/authorization/AuthorizationContextFactory.java b/src/main/java/iudx/rs/proxy/authenticator/authorization/AuthorizationContextFactory.java index b264cc5..3d807f5 100644 --- a/src/main/java/iudx/rs/proxy/authenticator/authorization/AuthorizationContextFactory.java +++ b/src/main/java/iudx/rs/proxy/authenticator/authorization/AuthorizationContextFactory.java @@ -4,12 +4,16 @@ public class AuthorizationContextFactory { private final static AuthorizationStrategy consumerAuth = new ConsumerAuthStrategy(); + private final static AuthorizationStrategy providerAuth = new ProviderAuthStrategy(); public static AuthorizationStrategy create(IudxRole role) { switch (role) { case CONSUMER: { return consumerAuth; } + case PROVIDER: { + return providerAuth; + } default: throw new IllegalArgumentException(role + "role is not defined in IUDX"); } diff --git a/src/main/java/iudx/rs/proxy/authenticator/authorization/ConsumerAuthStrategy.java b/src/main/java/iudx/rs/proxy/authenticator/authorization/ConsumerAuthStrategy.java index 61089cd..a11d6ae 100644 --- a/src/main/java/iudx/rs/proxy/authenticator/authorization/ConsumerAuthStrategy.java +++ b/src/main/java/iudx/rs/proxy/authenticator/authorization/ConsumerAuthStrategy.java @@ -1,8 +1,8 @@ package iudx.rs.proxy.authenticator.authorization; -import static iudx.rs.proxy.apiserver.util.RequestType.ENTITY; import static iudx.rs.proxy.authenticator.authorization.Api.ENTITIES; import static iudx.rs.proxy.authenticator.authorization.Api.TEMPORAL; +import static iudx.rs.proxy.authenticator.authorization.Api.CONSUMER_AUDIT; import static iudx.rs.proxy.authenticator.authorization.Method.GET; import io.vertx.core.json.JsonArray; @@ -23,6 +23,7 @@ public class ConsumerAuthStrategy implements AuthorizationStrategy { // api access list/rules List apiAccessList = new ArrayList<>(); apiAccessList.add(new AuthorizationRequest(GET, TEMPORAL)); + apiAccessList.add(new AuthorizationRequest(GET, CONSUMER_AUDIT)); apiAccessList.add(new AuthorizationRequest(GET, ENTITIES)); consumerAuthorizationRules.put(IudxAccess.API.getAccess(), apiAccessList); diff --git a/src/main/java/iudx/rs/proxy/authenticator/authorization/IudxRole.java b/src/main/java/iudx/rs/proxy/authenticator/authorization/IudxRole.java index ac427ff..865a458 100644 --- a/src/main/java/iudx/rs/proxy/authenticator/authorization/IudxRole.java +++ b/src/main/java/iudx/rs/proxy/authenticator/authorization/IudxRole.java @@ -4,7 +4,8 @@ public enum IudxRole { - CONSUMER("consumer"); + CONSUMER("consumer"), + PROVIDER("provider"); private final String role; diff --git a/src/main/java/iudx/rs/proxy/authenticator/authorization/ProviderAuthStrategy.java b/src/main/java/iudx/rs/proxy/authenticator/authorization/ProviderAuthStrategy.java new file mode 100644 index 0000000..de8fa28 --- /dev/null +++ b/src/main/java/iudx/rs/proxy/authenticator/authorization/ProviderAuthStrategy.java @@ -0,0 +1,24 @@ +package iudx.rs.proxy.authenticator.authorization; + +import iudx.rs.proxy.authenticator.model.JwtData; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +public class ProviderAuthStrategy implements AuthorizationStrategy { + + private static final Logger LOGGER = LogManager.getLogger(ProviderAuthStrategy.class); + + static Map> providerAuthorizationRules = new HashMap<>(); + + static { + // provider allowed to access all endpoints + } + + @Override + public boolean isAuthorized(AuthorizationRequest authRequest, JwtData jwtData) { + return true; + } +} diff --git a/src/main/java/iudx/rs/proxy/common/ResponseUrn.java b/src/main/java/iudx/rs/proxy/common/ResponseUrn.java index eba371c..2f935f8 100644 --- a/src/main/java/iudx/rs/proxy/common/ResponseUrn.java +++ b/src/main/java/iudx/rs/proxy/common/ResponseUrn.java @@ -5,7 +5,7 @@ public enum ResponseUrn { SUCCESS_URN("urn:dx:rs:success", "successful operations"), - INVALID_PARAM_URN("urn:dx:rs:invalidParamameter", "Invalid parameter passed"), + INVALID_PARAM_URN("urn:dx:rs:invalidParameter", "Invalid parameter passed"), INVALID_GEO_REL_URN("urn:dx:rs:invalidGeoRel", "Invalid geo relation value"), INVALID_TEMPORAL_PARAM_URN("urn:dx:rs:invalidTemporalParam", "Invalid temporal parameter"), INVALID_TEMPORAL_REL_URN("urn:dx:rs:invalidTemporalRelationParam", "Invalid temporal param value"), @@ -29,8 +29,9 @@ public enum ResponseUrn { // extra urn INVALID_ID_VALUE_URN("urn:dx:rs:invalidIdValue", "Invalid id"), + INVALID_PROVIDER_ID_VALUE_URN("urn:dx:rs:invalidIdValue", "Invalid id"), INVALID_PAYLOAD_FORMAT_URN("urn:dx:rs:invalidPayloadFormat", "Invalid json format in post request [schema mismatch]"), - INVALID_PARAM_VALUE_URN("urn:dx:rs:invalidParamameterValue", "Invalid parameter value passed"), + INVALID_PARAM_VALUE_URN("urn:dx:rs:invalidParameterValue", "Invalid parameter value passed"), BAD_REQUEST_URN("urn:dx:rs:badRequest","bad request parameter"), INVALID_HEADER_VALUE_URN("urn:dx:rs:invalidHeaderValue","Invalid header value"), DB_ERROR_URN("urn:dx:rs:DatabaseError","Database error"), diff --git a/src/main/java/iudx/rs/proxy/metering/MeteringServiceImpl.java b/src/main/java/iudx/rs/proxy/metering/MeteringServiceImpl.java index 7a92ef8..3959006 100644 --- a/src/main/java/iudx/rs/proxy/metering/MeteringServiceImpl.java +++ b/src/main/java/iudx/rs/proxy/metering/MeteringServiceImpl.java @@ -2,7 +2,9 @@ import static iudx.rs.proxy.metering.util.Constants.API; import static iudx.rs.proxy.metering.util.Constants.API_COLUMN; +import static iudx.rs.proxy.metering.util.Constants.BETWEEN; import static iudx.rs.proxy.metering.util.Constants.CONSUMER; +import static iudx.rs.proxy.metering.util.Constants.COUNT; import static iudx.rs.proxy.metering.util.Constants.COUNT_COLUMN; import static iudx.rs.proxy.metering.util.Constants.DATABASE_IP; import static iudx.rs.proxy.metering.util.Constants.DATABASE_NAME; @@ -14,25 +16,30 @@ import static iudx.rs.proxy.metering.util.Constants.ENDPOINT; import static iudx.rs.proxy.metering.util.Constants.END_TIME; import static iudx.rs.proxy.metering.util.Constants.ERROR; -import static iudx.rs.proxy.metering.util.Constants.FAILED; import static iudx.rs.proxy.metering.util.Constants.HEADER_OPTIONS; import static iudx.rs.proxy.metering.util.Constants.ID; +import static iudx.rs.proxy.metering.util.Constants.ID_COLUMN; import static iudx.rs.proxy.metering.util.Constants.INVALID_PROVIDER_REQUIRED; import static iudx.rs.proxy.metering.util.Constants.IUDX_PROVIDER_AUDIT_URL; -import static iudx.rs.proxy.metering.util.Constants.MESSAGE; +import static iudx.rs.proxy.metering.util.Constants.LAST_ID; +import static iudx.rs.proxy.metering.util.Constants.LATEST_ID; import static iudx.rs.proxy.metering.util.Constants.POOL_SIZE; import static iudx.rs.proxy.metering.util.Constants.PROVIDER_ID; import static iudx.rs.proxy.metering.util.Constants.QUERY_KEY; -import static iudx.rs.proxy.metering.util.Constants.RESOURCEID_COLUMN; +import static iudx.rs.proxy.metering.util.Constants.RESOURCE_ID_COLUMN; +import static iudx.rs.proxy.metering.util.Constants.RESPONSE_ARRAY; +import static iudx.rs.proxy.metering.util.Constants.RESPONSE_LIMIT_EXCEED; import static iudx.rs.proxy.metering.util.Constants.RESPONSE_SIZE_COLUMN; +import static iudx.rs.proxy.metering.util.Constants.RESULTS; import static iudx.rs.proxy.metering.util.Constants.START_TIME; -import static iudx.rs.proxy.metering.util.Constants.SUCCESS; import static iudx.rs.proxy.metering.util.Constants.TIME; import static iudx.rs.proxy.metering.util.Constants.TIME_COLUMN; import static iudx.rs.proxy.metering.util.Constants.TIME_NOT_FOUND; import static iudx.rs.proxy.metering.util.Constants.TIME_RELATION; import static iudx.rs.proxy.metering.util.Constants.TIME_RELATION_NOT_FOUND; -import static iudx.rs.proxy.metering.util.Constants.TOTAL; +import static iudx.rs.proxy.metering.util.Constants.TITLE; +import static iudx.rs.proxy.metering.util.Constants.TOTAL_HITS; +import static iudx.rs.proxy.metering.util.Constants.TYPE_KEY; import static iudx.rs.proxy.metering.util.Constants.USERID_COLUMN; import static iudx.rs.proxy.metering.util.Constants.USERID_NOT_FOUND; import static iudx.rs.proxy.metering.util.Constants.USER_ID; @@ -49,8 +56,10 @@ import io.vertx.sqlclient.PoolOptions; import io.vertx.sqlclient.Row; import io.vertx.sqlclient.RowSet; +import iudx.rs.proxy.common.Response; +import iudx.rs.proxy.common.ResponseUrn; import iudx.rs.proxy.metering.util.QueryBuilder; -import iudx.rs.proxy.metering.util.ResponseBuilder; +import org.apache.http.HttpStatus; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -63,6 +72,7 @@ public class MeteringServiceImpl implements MeteringService { public final String _USERID_COLUMN; public final String _TIME_COLUMN; public final String _RESPONSE_SIZE_COLUMN; + public final String _ID_COLUMN; private final Vertx vertx; private final QueryBuilder queryBuilder = new QueryBuilder(); PgConnectOptions connectOptions; @@ -74,9 +84,8 @@ public class MeteringServiceImpl implements MeteringService { private String databaseName; private String databaseUserName; private String databasePassword; - private String databaseTableName; private int databasePoolSize; - private ResponseBuilder responseBuilder; + private String databaseTableName; public MeteringServiceImpl(JsonObject propObj, Vertx vertxInstance) { @@ -102,7 +111,7 @@ public MeteringServiceImpl(JsonObject propObj, Vertx vertxInstance) { _COUNT_COLUMN = COUNT_COLUMN.insert(0, "(" + databaseName + "." + databaseTableName + ".").toString(); _RESOURCE_ID_COLUMN = - RESOURCEID_COLUMN.insert(0, "(" + databaseName + "." + databaseTableName + ".").toString(); + RESOURCE_ID_COLUMN.insert(0, "(" + databaseName + "." + databaseTableName + ".").toString(); _API_COLUMN = API_COLUMN.insert(0, "(" + databaseName + "." + databaseTableName + ".").toString(); _USERID_COLUMN = @@ -112,131 +121,193 @@ public MeteringServiceImpl(JsonObject propObj, Vertx vertxInstance) { _RESPONSE_SIZE_COLUMN = RESPONSE_SIZE_COLUMN.insert(0, "(" + databaseName + "." + databaseTableName + ".") .toString(); + _ID_COLUMN = ID_COLUMN.insert(0, "(" + databaseName + "." + databaseTableName + ".").toString(); } @Override public MeteringService executeReadQuery(JsonObject request, Handler> handler) { - LOGGER.trace("Info: Count Query" + request.toString()); + LOGGER.trace("Info: Read Query" + request.toString()); if (request.getString(ENDPOINT).equals(IUDX_PROVIDER_AUDIT_URL) && request.getString(PROVIDER_ID) == null) { - responseBuilder = - new ResponseBuilder(FAILED).setTypeAndTitle(400).setMessage(INVALID_PROVIDER_REQUIRED); - handler.handle(Future.failedFuture(responseBuilder.getResponse().toString())); + Response response = + new Response.Builder().withUrn(ResponseUrn.INVALID_PROVIDER_ID_VALUE_URN.getUrn()) + .withStatus(HttpStatus.SC_BAD_REQUEST).withDetail(INVALID_PROVIDER_REQUIRED).build(); + handler.handle(Future.failedFuture(response.toString())); return this; } if (request.getString(TIME_RELATION) == null || - !request.getString(TIME_RELATION).equals(DURING)) { + !(request.getString(TIME_RELATION).equals(DURING) || + request.getString(TIME_RELATION).equals(BETWEEN))) { LOGGER.debug("Info: " + TIME_RELATION_NOT_FOUND); - responseBuilder = - new ResponseBuilder(FAILED).setTypeAndTitle(400).setMessage(TIME_RELATION_NOT_FOUND); - handler.handle(Future.failedFuture(responseBuilder.getResponse().toString())); + Response response = + new Response.Builder().withUrn(ResponseUrn.INVALID_TEMPORAL_REL_URN.getUrn()) + .withStatus(HttpStatus.SC_BAD_REQUEST).withDetail(TIME_RELATION_NOT_FOUND).build(); + handler.handle(Future.failedFuture(response.toString())); return this; } if (request.getString(START_TIME) == null || request.getString(END_TIME) == null) { LOGGER.debug("Info: " + TIME_NOT_FOUND); - responseBuilder = new ResponseBuilder(FAILED).setTypeAndTitle(400).setMessage(TIME_NOT_FOUND); - handler.handle(Future.failedFuture(responseBuilder.getResponse().toString())); + Response response = + new Response.Builder().withUrn(ResponseUrn.INVALID_TEMPORAL_PARAM_URN.getUrn()) + .withStatus(HttpStatus.SC_BAD_REQUEST).withDetail(TIME_NOT_FOUND).build(); + handler.handle(Future.failedFuture(response.toString())); return this; } if (request.getString(USER_ID) == null || request.getString(USER_ID).isEmpty()) { LOGGER.debug("Info: " + USERID_NOT_FOUND); - responseBuilder = - new ResponseBuilder(FAILED).setTypeAndTitle(400).setMessage(USERID_NOT_FOUND); - handler.handle(Future.failedFuture(responseBuilder.getResponse().toString())); + Response response = new Response.Builder().withUrn(ResponseUrn.INVALID_ID_VALUE_URN.getUrn()) + .withStatus(HttpStatus.SC_BAD_REQUEST).withDetail(USERID_NOT_FOUND).build(); + handler.handle(Future.failedFuture(response.toString())); return this; } request.put(DATABASE_TABLE_NAME, databaseTableName); - query = queryBuilder.buildReadingQuery(request); - System.out.println("query-----"+query); + query = queryBuilder.buildCountQuery(request); + LOGGER.trace(query); if (query.containsKey(ERROR)) { LOGGER.error("Fail: Query returned with an error: " + query.getString(ERROR)); - responseBuilder = - new ResponseBuilder(FAILED).setTypeAndTitle(400).setMessage(query.getString(ERROR)); - handler.handle(Future.failedFuture(responseBuilder.getResponse().toString())); + Response response = + new Response.Builder().withUrn(ResponseUrn.INVALID_PARAM_VALUE_URN.getUrn()) + .withStatus(HttpStatus.SC_BAD_REQUEST).withDetail(query.getString(ERROR)).build(); + handler.handle(Future.failedFuture(response.toString())); return this; } LOGGER.debug("Info: Query constructed: " + query.getString(QUERY_KEY)); - Future result; - if (request.getString(HEADER_OPTIONS) != null) { - result = executeCountQuery(query); - } else { - result = executeReadQuery(query); - } + Future countResult = executeCountQuery(query); + countResult.onComplete(countResultHandler -> { + if (countResultHandler.succeeded()) { + int totalCount = Integer.parseInt( + countResultHandler.result().getJsonArray(RESULTS).getJsonObject(0) + .getString(TOTAL_HITS)); + if (request.getString(HEADER_OPTIONS) != null) { + handler.handle(Future.succeededFuture(countResultHandler.result())); + } else { - result.onComplete(resultHandler -> { - if (resultHandler.succeeded()) { - handler.handle(Future.succeededFuture(resultHandler.result())); - } else if (resultHandler.failed()) { - LOGGER.error("Read from DB failed:" + resultHandler.cause()); - handler.handle(Future.failedFuture(resultHandler.cause().getMessage())); + if (totalCount >= 10000 || totalCount == 0) { + Response response = + new Response.Builder().withUrn(ResponseUrn.INVALID_PARAM_VALUE_URN.getUrn()) + .withStatus(HttpStatus.SC_BAD_REQUEST).withDetail(RESPONSE_LIMIT_EXCEED) + .build(); + handler.handle(Future.failedFuture(response.toString())); + } else { + query = queryBuilder.buildReadingQuery(request); + query.put(COUNT, totalCount); + Future initialReadResult = executeReadQuery(query); + + initialReadResult.onComplete(initialReadHandler -> { + if (initialReadHandler.succeeded()) { + Future remainingReadResult = executeRemainingReadQuery(query); + remainingReadResult.onComplete(remainingReadHandler -> { + if (remainingReadHandler.succeeded()) { + JsonArray jsonArray = + remainingReadHandler.result().getJsonArray(RESPONSE_ARRAY); + JsonObject response = + new JsonObject().put(TYPE_KEY, ResponseUrn.SUCCESS_URN.getUrn()) + .put(TITLE, ResponseUrn.SUCCESS_URN.getMessage()) + .put(TOTAL_HITS, totalCount).put(RESULTS, jsonArray); + handler.handle(Future.succeededFuture(response)); + } else { + LOGGER.info("FAILED " + remainingReadHandler.cause()); + } + }); + } + }); + } + } + } else if (countResultHandler.failed()) { + LOGGER.error("Read from DB failed:" + countResultHandler.cause()); + handler.handle(Future.failedFuture(countResultHandler.cause().getMessage())); } }); return this; } + /* First time read query */ private Future executeReadQuery(JsonObject query) { Promise promise = Promise.promise(); - JsonObject response = new JsonObject(); + JsonArray jsonArray = new JsonArray(); pool.withConnection(connection -> connection.query(query.getString(QUERY_KEY)).execute()) .onSuccess(rows -> { - JsonArray jsonArray = new JsonArray(); - RowSet result = rows; - for (Row rs : result) { + String lastId = ""; + for (Row rs : rows) { JsonObject temp = new JsonObject(); temp.put(ID, rs.getString(_RESOURCE_ID_COLUMN)); temp.put(TIME, rs.getString(_TIME_COLUMN)); temp.put(API, rs.getString(_API_COLUMN)); temp.put(CONSUMER, rs.getString(_USERID_COLUMN)); + lastId = rs.getString(_ID_COLUMN); jsonArray.add(temp); } - - if (jsonArray.isEmpty()) { - responseBuilder = new ResponseBuilder(FAILED).setTypeAndTitle(204); - promise.fail(responseBuilder.getResponse().toString()); - } else { - responseBuilder = new ResponseBuilder(SUCCESS).setTypeAndTitle(200).setData(jsonArray); - promise.complete(responseBuilder.getResponse()); - } + query.put(LAST_ID, lastId); + query.put(LATEST_ID, lastId); + query.put(RESPONSE_ARRAY, jsonArray); + promise.complete(query); }).onFailure(event -> { promise.fail("Failed to get connection from the database"); }); - return promise.future(); } private Future executeCountQuery(JsonObject query) { Promise promise = Promise.promise(); - JsonObject response = new JsonObject(); pool.withConnection(connection -> connection.query(query.getString(QUERY_KEY)).execute()) .onSuccess(rows -> { - RowSet result = rows; - for (Row rs : result) { + int count = 0; + for (Row rs : rows) { LOGGER.debug("COUNT: " + (rs.getInteger(_COUNT_COLUMN))); - response.put(TOTAL, rs.getInteger(_COUNT_COLUMN)); - } - if (response.getInteger(TOTAL) == 0) { - responseBuilder = new ResponseBuilder(FAILED).setTypeAndTitle(204); - promise.fail(responseBuilder.getResponse().toString()); - } else { - responseBuilder = new ResponseBuilder(SUCCESS).setTypeAndTitle(200) - .setCount(response.getInteger(TOTAL)); - LOGGER.debug("Info: " + responseBuilder.getResponse().toString()); - promise.complete(responseBuilder.getResponse()); + count = rs.getInteger(_COUNT_COLUMN); } + JsonObject response = new JsonObject().put(TYPE_KEY, ResponseUrn.SUCCESS_URN.getUrn()) + .put(TITLE, ResponseUrn.SUCCESS_URN.getMessage()) + .put(RESULTS, new JsonArray().add(new JsonObject().put(TOTAL_HITS, count))); + promise.complete(response); }).onFailure(event -> { promise.fail("Failed to get connection from the database"); }); return promise.future(); } + private Future executeRemainingReadQuery(JsonObject query) { + final Promise promise = Promise.promise(); + executeRemainingReadQuery(promise, query); + return promise.future(); + } + + private void executeRemainingReadQuery(final Promise promise, + final JsonObject query) { + String latestId = query.getString(LATEST_ID); + JsonArray jsonArray = query.getJsonArray(RESPONSE_ARRAY); + if (latestId.isEmpty()) { + promise.complete(query); + } else { + String tempQuery = queryBuilder.buildTempReadQuery(query); + query.put(QUERY_KEY, tempQuery); + query.put(LAST_ID, query.getValue(LATEST_ID)); + pool.withConnection(connection -> connection.query(tempQuery).execute()).onComplete(rows -> { + String currId = ""; + RowSet result = rows.result(); + for (Row rs : result) { + JsonObject temp = new JsonObject(); + temp.put(ID, rs.getString(_RESOURCE_ID_COLUMN)); + temp.put(TIME, rs.getString(_TIME_COLUMN)); + temp.put(API, rs.getString(_API_COLUMN)); + temp.put(CONSUMER, rs.getString(_USERID_COLUMN)); + currId = rs.getString(_ID_COLUMN); + jsonArray.add(temp); + } + query.put(LATEST_ID, currId); + executeRemainingReadQuery(promise, query); + }); + } + } + @Override public MeteringService executeWriteQuery(JsonObject request, Handler> handler) { @@ -256,25 +327,22 @@ public MeteringService executeWriteQuery(JsonObject request, private Future writeInDatabase(JsonObject query) { Promise promise = Promise.promise(); - JsonObject response = new JsonObject(); pool.withConnection(connection -> connection.query(query.getString(QUERY_KEY)).execute()) .onComplete(rows -> { if (rows.succeeded()) { - response.put(MESSAGE, "Table Updated Successfully"); - responseBuilder = new ResponseBuilder(SUCCESS).setTypeAndTitle(200) - .setMessage(response.getString(MESSAGE)); - LOGGER.debug("Info: " + responseBuilder.getResponse().toString()); - promise.complete(responseBuilder.getResponse()); + JsonObject response = new JsonObject().put(TYPE_KEY, ResponseUrn.SUCCESS_URN.getUrn()) + .put(TITLE, ResponseUrn.SUCCESS_URN.getMessage()); + promise.complete(response); } if (rows.failed()) { LOGGER.error("Info: failed :" + rows.cause()); - response.put(MESSAGE, rows.cause().getMessage()); - responseBuilder = new ResponseBuilder(FAILED).setTypeAndTitle(400) - .setMessage(response.getString(MESSAGE)); - LOGGER.info("Info: " + responseBuilder.getResponse().toString()); - promise.fail(responseBuilder.getResponse().toString()); + Response response = new Response.Builder().withUrn(ResponseUrn.BAD_REQUEST_URN.getUrn()) + .withStatus(HttpStatus.SC_BAD_REQUEST).withDetail(rows.cause().getMessage()) + .build(); + LOGGER.info("Info: " + response); + promise.fail(response.toString()); } }); return promise.future(); } -} +} \ No newline at end of file diff --git a/src/main/java/iudx/rs/proxy/metering/util/Constants.java b/src/main/java/iudx/rs/proxy/metering/util/Constants.java index 270b9ca..2c57630 100644 --- a/src/main/java/iudx/rs/proxy/metering/util/Constants.java +++ b/src/main/java/iudx/rs/proxy/metering/util/Constants.java @@ -5,8 +5,9 @@ public class Constants { public static final String ID = "id"; /* Temporal */ public static final String START_TIME = "startTime"; - public static final String TIME="time"; + public static final String TIME = "time"; public static final String END_TIME = "endTime"; + public static final String BETWEEN = "between"; public static final String TIME_RELATION = "timeRelation"; public static final String DURING = "during"; public static final String HEADER_OPTIONS = "options"; @@ -14,39 +15,33 @@ public class Constants { public static final String IUDX_PROVIDER_AUDIT_URL = IUDX_ADAPTOR_URL + "/provider/audit"; /* configs */ - public static final String DATABASE_IP="meteringDatabaseIP"; - public static final String DATABASE_PORT="meteringDatabasePort"; - public static final String DATABASE_NAME="meteringDatabaseName"; - public static final String DATABASE_USERNAME="meteringDatabaseUserName"; - public static final String DATABASE_PASSWORD="meteringDatabasePassword"; - public static final String DATABASE_TABLE_NAME="meteringDatabaseTableName"; - public static final String POOL_SIZE="meteringPoolSize"; - + public static final String DATABASE_IP = "meteringDatabaseIP"; + public static final String DATABASE_PORT = "meteringDatabasePort"; + public static final String DATABASE_NAME = "meteringDatabaseName"; + public static final String DATABASE_USERNAME = "meteringDatabaseUserName"; + public static final String DATABASE_PASSWORD = "meteringDatabasePassword"; + public static final String DATABASE_TABLE_NAME = "meteringDatabaseTableName"; + public static final String POOL_SIZE = "meteringPoolSize"; /* Errors */ - public static final String SUCCESS = "Success"; - public static final String FAILED = "Failed"; - public static final String EMPTY_RESPONSE = "Empty response"; + public static final String SUCCESS = "successful operations"; public static final String DETAIL = "detail"; public static final String TITLE = "title"; public static final String RESULTS = "results"; - public static final String STATUS = "status"; - public static final String INVALID_RESOURCE_ID = "Invalid resource id"; - public static final String ROOT_CAUSE = "root_cause"; - public static final String REASON = "reason"; - /* Database */ public static final String ERROR = "Error"; public static final String QUERY_KEY = "query"; - public static final String TOTAL = "total"; public static final String TYPE_KEY = "type"; public static final String PROVIDER_ID = "providerID"; public static final String CONSUMER_ID = "consumerID"; public static final String ENDPOINT = "endPoint"; public static final String IID = "iid"; public static final String RESOURCE_ID = "resourceId"; - public static final String RESPONSE_SIZE="response_size"; + public static final String RESPONSE_SIZE = "response_size"; + public static final String LATEST_ID = "latestId"; + public static final String LAST_ID = "lastId"; + public static final String WHERE = "where"; /* Metering Service Constants*/ public static final String TIME_RELATION_NOT_FOUND = "Time relation not found."; @@ -57,35 +52,38 @@ public class Constants { public static final String INVALID_PROVIDER_REQUIRED = "provider id required."; public static final String INVALID_DATE_DIFFERENCE = "Difference between dates cannot be greater than 14 days or less than zero day."; - public static final String RESOURCE_QUERY = " and resourceId='$4';"; + public static final String RESOURCE_QUERY = " and resourceId='$4'"; - public static final String CONSUMERID_TIME_INTERVAL_COUNT_QUERY = + public static final String CONSUMER_ID_TIME_INTERVAL_COUNT_QUERY = "SELECT count(*) FROM $0 where epochtime>=$1 and epochtime<=$2 and userid='$3'"; - public static final String PROVIDERID_TIME_INTERVAL_COUNT_QUERY = + public static final String PROVIDER_ID_TIME_INTERVAL_COUNT_QUERY = "SELECT count(*) FROM $0 where epochtime>=$1 and epochtime<=$2 and providerid='$3'"; - public static final String CONSUMERID_TIME_INTERVAL_READ_QUERY = + public static final String CONSUMER_ID_TIME_INTERVAL_READ_QUERY = "SELECT * FROM $0 where epochtime>=$1 and epochtime<=$2 and userid='$3'"; - public static final String PROVIDERID_TIME_INTERVAL_READ_QUERY = + public static final String PROVIDER_ID_TIME_INTERVAL_READ_QUERY = "SELECT * FROM $0 where epochtime>=$1 and epochtime<=$2 and providerid='$3'"; - + public static final String ORDER_BY_AND_LIMIT = " ORDER BY ID LIMIT 999;"; public static final String API_QUERY = " and api='$5'"; public static final String USER_ID_QUERY = " and userid='$6'"; + public static final String ID_QUERY = " id>'$7' and"; public static final String API = "api"; public static final String USER_ID = "userid"; public static final String CONSUMER = "consumer"; - + public static final String COUNT = "count"; + public static final String RESPONSE_ARRAY = "responseArray"; + public static final String RESPONSE_LIMIT_EXCEED = "Requested time range exceeds response limit"; + public static final String TOTAL_HITS = "totalHits"; public static final String WRITE_QUERY = "INSERT INTO $0 (id,api,userid,epochtime,resourceid,isotime,providerid,size) VALUES ('$1','$2','$3',$4,'$5','$6','$7',$8)"; public static final StringBuilder COUNT_COLUMN = new StringBuilder("col0)"); - public static final StringBuilder RESOURCEID_COLUMN = new StringBuilder("resourceid)"); + public static final StringBuilder RESOURCE_ID_COLUMN = new StringBuilder("resourceid)"); public static final StringBuilder API_COLUMN = new StringBuilder("api)"); public static final StringBuilder USERID_COLUMN = new StringBuilder("userid)"); public static final StringBuilder TIME_COLUMN = new StringBuilder("isotime)"); public static final StringBuilder RESPONSE_SIZE_COLUMN = new StringBuilder("size)"); - - public static final String MESSAGE = "message"; + public static final StringBuilder ID_COLUMN = new StringBuilder("id)"); } diff --git a/src/main/java/iudx/rs/proxy/metering/util/QueryBuilder.java b/src/main/java/iudx/rs/proxy/metering/util/QueryBuilder.java index 2358d32..4e8f628 100644 --- a/src/main/java/iudx/rs/proxy/metering/util/QueryBuilder.java +++ b/src/main/java/iudx/rs/proxy/metering/util/QueryBuilder.java @@ -2,28 +2,31 @@ import static iudx.rs.proxy.metering.util.Constants.API; import static iudx.rs.proxy.metering.util.Constants.API_QUERY; -import static iudx.rs.proxy.metering.util.Constants.CONSUMERID_TIME_INTERVAL_COUNT_QUERY; -import static iudx.rs.proxy.metering.util.Constants.CONSUMERID_TIME_INTERVAL_READ_QUERY; +import static iudx.rs.proxy.metering.util.Constants.CONSUMER_ID_TIME_INTERVAL_COUNT_QUERY; +import static iudx.rs.proxy.metering.util.Constants.CONSUMER_ID_TIME_INTERVAL_READ_QUERY; import static iudx.rs.proxy.metering.util.Constants.CONSUMER_ID; import static iudx.rs.proxy.metering.util.Constants.DATABASE_TABLE_NAME; import static iudx.rs.proxy.metering.util.Constants.END_TIME; import static iudx.rs.proxy.metering.util.Constants.ERROR; -import static iudx.rs.proxy.metering.util.Constants.HEADER_OPTIONS; import static iudx.rs.proxy.metering.util.Constants.ID; +import static iudx.rs.proxy.metering.util.Constants.ID_QUERY; import static iudx.rs.proxy.metering.util.Constants.IID; import static iudx.rs.proxy.metering.util.Constants.INVALID_DATE_DIFFERENCE; import static iudx.rs.proxy.metering.util.Constants.INVALID_DATE_TIME; import static iudx.rs.proxy.metering.util.Constants.INVALID_PROVIDER_ID; -import static iudx.rs.proxy.metering.util.Constants.PROVIDERID_TIME_INTERVAL_COUNT_QUERY; -import static iudx.rs.proxy.metering.util.Constants.PROVIDERID_TIME_INTERVAL_READ_QUERY; +import static iudx.rs.proxy.metering.util.Constants.LAST_ID; +import static iudx.rs.proxy.metering.util.Constants.LATEST_ID; +import static iudx.rs.proxy.metering.util.Constants.ORDER_BY_AND_LIMIT; +import static iudx.rs.proxy.metering.util.Constants.PROVIDER_ID_TIME_INTERVAL_COUNT_QUERY; +import static iudx.rs.proxy.metering.util.Constants.PROVIDER_ID_TIME_INTERVAL_READ_QUERY; import static iudx.rs.proxy.metering.util.Constants.PROVIDER_ID; import static iudx.rs.proxy.metering.util.Constants.QUERY_KEY; import static iudx.rs.proxy.metering.util.Constants.RESOURCE_ID; import static iudx.rs.proxy.metering.util.Constants.RESOURCE_QUERY; -import static iudx.rs.proxy.metering.util.Constants.RESPONSE_SIZE; import static iudx.rs.proxy.metering.util.Constants.START_TIME; import static iudx.rs.proxy.metering.util.Constants.USER_ID; import static iudx.rs.proxy.metering.util.Constants.USER_ID_QUERY; +import static iudx.rs.proxy.metering.util.Constants.WHERE; import static iudx.rs.proxy.metering.util.Constants.WRITE_QUERY; import io.vertx.core.json.JsonObject; @@ -53,7 +56,7 @@ public JsonObject buildReadingQuery(JsonObject request) { String databaseTableName = request.getString(DATABASE_TABLE_NAME); StringBuilder query, tempQuery; - if (providerID != null && !checkProviderId(iid, providerID)) { + if (providerID != null && checkProviderId(iid, providerID)) { return new JsonObject().put(ERROR, INVALID_PROVIDER_ID); } /* check if the time is valid based on ISO 8601 format. */ @@ -73,10 +76,10 @@ public JsonObject buildReadingQuery(JsonObject request) { LOGGER.trace( "PERIOD between given time " - + (zonedDateTimeDifference(startZDT, endZDT, ChronoUnit.DAYS))); + + (zonedDateTimeDifference(startZDT, endZDT))); - if (zonedDateTimeDifference(startZDT, endZDT, ChronoUnit.DAYS) > 14 - || zonedDateTimeDifference(startZDT, endZDT, ChronoUnit.DAYS) <= 0) { + if (zonedDateTimeDifference(startZDT, endZDT) > 14 + || zonedDateTimeDifference(startZDT, endZDT) <= 0) { LOGGER.error(INVALID_DATE_DIFFERENCE); return new JsonObject().put(ERROR, INVALID_DATE_DIFFERENCE); } @@ -86,42 +89,109 @@ public JsonObject buildReadingQuery(JsonObject request) { long toTime = getEpochTime(endZDT); - if (request.getString(HEADER_OPTIONS) != null) { - if (providerID != null) { - query = - new StringBuilder( - PROVIDERID_TIME_INTERVAL_COUNT_QUERY - .replace("$0", databaseTableName) - .replace("$1", Long.toString(fromTime)) - .replace("$2", Long.toString(toTime)) - .replace("$3", providerID)); - } else { - query = - new StringBuilder( - CONSUMERID_TIME_INTERVAL_COUNT_QUERY - .replace("$0", databaseTableName) - .replace("$1", Long.toString(fromTime)) - .replace("$2", Long.toString(toTime)) - .replace("$3", userId)); - } + if (providerID != null) { + query = + new StringBuilder( + PROVIDER_ID_TIME_INTERVAL_READ_QUERY + .replace("$0", databaseTableName) + .replace("$1", Long.toString(fromTime)) + .replace("$2", Long.toString(toTime)) + .replace("$3", providerID)); } else { - if (providerID != null){ - query = - new StringBuilder( - PROVIDERID_TIME_INTERVAL_READ_QUERY - .replace("$0", databaseTableName) - .replace("$1", Long.toString(fromTime)) - .replace("$2", Long.toString(toTime)) - .replace("$3", providerID)); - } else { - query = - new StringBuilder( - CONSUMERID_TIME_INTERVAL_READ_QUERY - .replace("$0", databaseTableName) - .replace("$1", Long.toString(fromTime)) - .replace("$2", Long.toString(toTime)) - .replace("$3", userId)); + query = + new StringBuilder( + CONSUMER_ID_TIME_INTERVAL_READ_QUERY + .replace("$0", databaseTableName) + .replace("$1", Long.toString(fromTime)) + .replace("$2", Long.toString(toTime)) + .replace("$3", userId)); + } + if (consumerID != null) { + tempQuery = query; + tempQuery.append(USER_ID_QUERY.replace("$6", consumerID)); + } + if (api != null && resourceId != null) { + tempQuery = query; + for (String s : + Arrays.asList(API_QUERY.replace("$5", api), RESOURCE_QUERY.replace("$4", resourceId))) { + tempQuery.append(s); } + } else if (api != null) { + tempQuery = query; + tempQuery.append(API_QUERY.replace("$5", api)); + } else if (resourceId != null) { + tempQuery = query; + tempQuery.append(RESOURCE_QUERY.replace("$4", resourceId)); + } else { + tempQuery = query; + } + LOGGER.trace("Info: QUERY " + tempQuery); + tempQuery.append(ORDER_BY_AND_LIMIT); + return new JsonObject().put(QUERY_KEY, tempQuery); + } + + public JsonObject buildCountQuery(JsonObject request) { + String startTime = request.getString(START_TIME); + String endTime = request.getString(END_TIME); + String resourceId = request.getString(RESOURCE_ID); + String userId = request.getString(USER_ID); + String api = request.getString(API); + String providerID = request.getString(PROVIDER_ID); + String consumerID = request.getString(CONSUMER_ID); + String iid = request.getString(IID); + String databaseTableName = request.getString(DATABASE_TABLE_NAME); + + StringBuilder query, tempQuery; + + if (providerID != null && checkProviderId(iid, providerID)) { + return new JsonObject().put(ERROR, INVALID_PROVIDER_ID); + } + /* check if the time is valid based on ISO 8601 format. */ + ZonedDateTime zdt; + try { + zdt = ZonedDateTime.parse(startTime); + LOGGER.debug("Parsed time: " + zdt.toString()); + zdt = ZonedDateTime.parse(endTime); + LOGGER.debug("Parsed time: " + zdt.toString()); + } catch (DateTimeParseException e) { + LOGGER.error("Invalid Date exception: " + e.getMessage()); + return new JsonObject().put(ERROR, INVALID_DATE_TIME); + } + + ZonedDateTime startZDT = ZonedDateTime.parse(startTime); + ZonedDateTime endZDT = ZonedDateTime.parse(endTime); + + LOGGER.trace( + "PERIOD between given time " + + (zonedDateTimeDifference(startZDT, endZDT))); + + if (zonedDateTimeDifference(startZDT, endZDT) > 14 + || zonedDateTimeDifference(startZDT, endZDT) <= 0) { + LOGGER.error(INVALID_DATE_DIFFERENCE); + return new JsonObject().put(ERROR, INVALID_DATE_DIFFERENCE); + } + + long fromTime = getEpochTime(startZDT); + LOGGER.debug("Epoch fromTime: " + fromTime); + + long toTime = getEpochTime(endZDT); + + if (providerID != null) { + query = + new StringBuilder( + PROVIDER_ID_TIME_INTERVAL_COUNT_QUERY + .replace("$0", databaseTableName) + .replace("$1", Long.toString(fromTime)) + .replace("$2", Long.toString(toTime)) + .replace("$3", providerID)); + } else { + query = + new StringBuilder( + CONSUMER_ID_TIME_INTERVAL_COUNT_QUERY + .replace("$0", databaseTableName) + .replace("$1", Long.toString(fromTime)) + .replace("$2", Long.toString(toTime)) + .replace("$3", userId)); } if (consumerID != null) { tempQuery = query; @@ -131,17 +201,18 @@ public JsonObject buildReadingQuery(JsonObject request) { tempQuery = query; for (String s : Arrays.asList(API_QUERY.replace("$5", api), RESOURCE_QUERY.replace("$4", resourceId))) { - tempQuery = tempQuery.append(s); + tempQuery.append(s); } } else if (api != null) { tempQuery = query; - tempQuery = tempQuery.append(API_QUERY.replace("$5", api)); + tempQuery.append(API_QUERY.replace("$5", api)); } else if (resourceId != null) { tempQuery = query; - tempQuery = tempQuery.append(RESOURCE_QUERY.replace("$4", resourceId)); + tempQuery.append(RESOURCE_QUERY.replace("$4", resourceId)); } else { tempQuery = query; } + tempQuery.insert(tempQuery.length(), ";"); LOGGER.trace("Info: QUERY " + tempQuery); return new JsonObject().put(QUERY_KEY, tempQuery); } @@ -161,8 +232,9 @@ public JsonObject buildWritingQuery(JsonObject request) { .atZone(ZoneId.of("Asia/Kolkata")) .truncatedTo(ChronoUnit.SECONDS) .toString(); + long response_size = request.getLong(Constants.RESPONSE_SIZE); String databaseTableName = request.getString(DATABASE_TABLE_NAME); - Long responseSize = request.getLong(RESPONSE_SIZE); + StringBuilder query = new StringBuilder( WRITE_QUERY @@ -174,21 +246,33 @@ public JsonObject buildWritingQuery(JsonObject request) { .replace("$5", resourceId) .replace("$6", isoTime) .replace("$7", providerID) - .replace("$8", Long.toString(responseSize))); + .replace("$8", Long.toString(response_size))); LOGGER.trace("Info: Query " + query); return new JsonObject().put(QUERY_KEY, query); } - private long zonedDateTimeDifference(ZonedDateTime d1, ZonedDateTime d2, ChronoUnit unit) { - return unit.between(d1, d2); + public String buildTempReadQuery(JsonObject request) { + StringBuilder tempQuery = new StringBuilder(request.getString(QUERY_KEY)); + String lastId = request.getString(LAST_ID); + String latestId = request.getString(LATEST_ID); + if (tempQuery.toString().contains(lastId)) { + return tempQuery.toString().replace(lastId, latestId); + } else { + tempQuery.insert(tempQuery.indexOf(WHERE) + 5, ID_QUERY.replace("$7", lastId)); + } + return tempQuery.toString(); + } + + private long zonedDateTimeDifference(ZonedDateTime d1, ZonedDateTime d2) { + return ChronoUnit.DAYS.between(d1, d2); } private boolean checkProviderId(String iid, String providerID) { - return iid.substring(0, iid.indexOf('/', iid.indexOf('/') + 1)).equals(providerID); + return !iid.substring(0, iid.indexOf('/', iid.indexOf('/') + 1)).equals(providerID); } private long getEpochTime(ZonedDateTime time) { return time.toInstant().toEpochMilli(); } -} +} \ No newline at end of file diff --git a/src/main/java/iudx/rs/proxy/metering/util/ResponseBuilder.java b/src/main/java/iudx/rs/proxy/metering/util/ResponseBuilder.java deleted file mode 100644 index cb31b05..0000000 --- a/src/main/java/iudx/rs/proxy/metering/util/ResponseBuilder.java +++ /dev/null @@ -1,58 +0,0 @@ -package iudx.rs.proxy.metering.util; - -import static iudx.rs.proxy.metering.util.Constants.DETAIL; -import static iudx.rs.proxy.metering.util.Constants.RESULTS; -import static iudx.rs.proxy.metering.util.Constants.SUCCESS; -import static iudx.rs.proxy.metering.util.Constants.TITLE; -import static iudx.rs.proxy.metering.util.Constants.TOTAL; -import static iudx.rs.proxy.metering.util.Constants.TYPE_KEY; - -import io.vertx.core.json.JsonArray; -import io.vertx.core.json.JsonObject; -import iudx.rs.proxy.common.ResponseUrn; - -public class ResponseBuilder { - private final String status; - private final JsonObject response; - - /** Initialise the object with Success or Failure. */ - public ResponseBuilder(String status) { - this.status = status; - response = new JsonObject(); - } - - public ResponseBuilder setTypeAndTitle(int statusCode) { - - if (200 == statusCode) { - response.put(TYPE_KEY, ResponseUrn.SUCCESS_URN.getUrn()); - response.put(TITLE, SUCCESS); - } else if (204 == statusCode) { - response.put(TYPE_KEY, statusCode); - response.put(TITLE, SUCCESS); - } else { - response.put(TYPE_KEY, statusCode); - response.put(TITLE, ResponseUrn.BAD_REQUEST_URN.getUrn()); - } - return this; - } - - /** Overloaded methods for Error messages. */ - public ResponseBuilder setMessage(String error) { - response.put(DETAIL, error); - return this; - } - - public ResponseBuilder setCount(int count) { - response.put(RESULTS, new JsonArray().add(new JsonObject().put(TOTAL, count))); - return this; - } - - public ResponseBuilder setData(JsonArray jsonArray) { - response.put(RESULTS, jsonArray); - return this; - } - - public JsonObject getResponse() { - return response; - } -} diff --git a/src/test/java/iudx/rs/proxy/metering/MeteringServiceImplTest.java b/src/test/java/iudx/rs/proxy/metering/MeteringServiceImplTest.java index 3684a13..7b6fddf 100644 --- a/src/test/java/iudx/rs/proxy/metering/MeteringServiceImplTest.java +++ b/src/test/java/iudx/rs/proxy/metering/MeteringServiceImplTest.java @@ -1,23 +1,51 @@ package iudx.rs.proxy.metering; +import static iudx.rs.proxy.metering.util.Constants.API; +import static iudx.rs.proxy.metering.util.Constants.CONSUMER_ID; +import static iudx.rs.proxy.metering.util.Constants.DATABASE_IP; +import static iudx.rs.proxy.metering.util.Constants.DATABASE_NAME; +import static iudx.rs.proxy.metering.util.Constants.DATABASE_PASSWORD; +import static iudx.rs.proxy.metering.util.Constants.DATABASE_PORT; +import static iudx.rs.proxy.metering.util.Constants.DATABASE_TABLE_NAME; +import static iudx.rs.proxy.metering.util.Constants.DATABASE_USERNAME; +import static iudx.rs.proxy.metering.util.Constants.DETAIL; +import static iudx.rs.proxy.metering.util.Constants.DURING; +import static iudx.rs.proxy.metering.util.Constants.ENDPOINT; +import static iudx.rs.proxy.metering.util.Constants.END_TIME; +import static iudx.rs.proxy.metering.util.Constants.ID; +import static iudx.rs.proxy.metering.util.Constants.IID; +import static iudx.rs.proxy.metering.util.Constants.INVALID_DATE_DIFFERENCE; +import static iudx.rs.proxy.metering.util.Constants.INVALID_DATE_TIME; +import static iudx.rs.proxy.metering.util.Constants.INVALID_PROVIDER_ID; +import static iudx.rs.proxy.metering.util.Constants.INVALID_PROVIDER_REQUIRED; +import static iudx.rs.proxy.metering.util.Constants.POOL_SIZE; +import static iudx.rs.proxy.metering.util.Constants.PROVIDER_ID; +import static iudx.rs.proxy.metering.util.Constants.RESOURCE_ID; +import static iudx.rs.proxy.metering.util.Constants.RESPONSE_LIMIT_EXCEED; +import static iudx.rs.proxy.metering.util.Constants.RESPONSE_SIZE; +import static iudx.rs.proxy.metering.util.Constants.START_TIME; +import static iudx.rs.proxy.metering.util.Constants.SUCCESS; +import static iudx.rs.proxy.metering.util.Constants.TIME_NOT_FOUND; +import static iudx.rs.proxy.metering.util.Constants.TIME_RELATION; +import static iudx.rs.proxy.metering.util.Constants.TIME_RELATION_NOT_FOUND; +import static iudx.rs.proxy.metering.util.Constants.TITLE; +import static iudx.rs.proxy.metering.util.Constants.USERID_NOT_FOUND; +import static iudx.rs.proxy.metering.util.Constants.USER_ID; +import static org.junit.jupiter.api.Assertions.assertEquals; + import io.vertx.core.Vertx; import io.vertx.core.json.JsonObject; import io.vertx.junit5.VertxExtension; import io.vertx.junit5.VertxTestContext; import iudx.rs.proxy.configuration.Configuration; +import java.util.UUID; import org.apache.log4j.LogManager; import org.apache.log4j.Logger; import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; -import java.util.UUID; - -import static iudx.rs.proxy.metering.util.Constants.*; -import static org.junit.jupiter.api.Assertions.*; - @ExtendWith(VertxExtension.class) class MeteringServiceImplTest { @@ -41,13 +69,13 @@ static void startVertex(Vertx vertx, VertxTestContext vertxTestContext) { vertxObj = vertx; config = new Configuration(); JsonObject dbConfig = config.configLoader(4, vertx); - databaseIP = dbConfig.getString("meteringDatabaseIP"); - databasePort = dbConfig.getInteger("meteringDatabasePort"); - databaseName = dbConfig.getString("meteringDatabaseName"); - databaseUserName = dbConfig.getString("meteringDatabaseUserName"); - databasePassword = dbConfig.getString("meteringDatabasePassword"); - databasePoolSize = dbConfig.getInteger("meteringPoolSize"); - databaseTableName = dbConfig.getString("meteringDatabaseTableName"); + databaseIP = dbConfig.getString(DATABASE_IP); + databasePort = dbConfig.getInteger(DATABASE_PORT); + databaseName = dbConfig.getString(DATABASE_NAME); + databaseUserName = dbConfig.getString(DATABASE_USERNAME); + databasePassword = dbConfig.getString(DATABASE_PASSWORD); + databaseTableName = dbConfig.getString(DATABASE_TABLE_NAME); + databasePoolSize = dbConfig.getInteger(POOL_SIZE); meteringService = new MeteringServiceImpl(dbConfig, vertxObj); userId = UUID.randomUUID().toString(); id = "89a36273d77dac4cf38114fca1bbe64392547f86"; @@ -56,13 +84,12 @@ static void startVertex(Vertx vertx, VertxTestContext vertxTestContext) { private JsonObject readConsumerRequest() { JsonObject jsonObject = new JsonObject(); - jsonObject.put(USER_ID, "844e251b-574b-46e6-9247-f76f1f70a637"); - jsonObject.put(RESOURCE_ID, - "iisc.ac.in/89a36273d77dac4cf38114fca1bbe64392547f86/rs.iudx.io/pune-env-flood/FWR055"); - jsonObject.put(START_TIME, "2022-03-20T05:30:00+05:30[Asia/Kolkata]"); - jsonObject.put(END_TIME, "2022-03-30T02:00:00+05:30[Asia/Kolkata]"); + jsonObject.put(USER_ID, "15c7506f-c800-48d6-adeb-0542b03947c6"); + jsonObject.put(RESOURCE_ID, "15c7506f-c800-48d6-adeb-0542b03947c6/integration-test-alias/"); + jsonObject.put(START_TIME, "2022-05-29T05:30:00+05:30[Asia/Kolkata]"); + jsonObject.put(END_TIME, "2022-06-04T02:00:00+05:30[Asia/Kolkata]"); jsonObject.put(TIME_RELATION, DURING); - jsonObject.put(API, "/ngsi-ld/v1/entities"); + jsonObject.put(API, "/ngsi-ld/v1/subscription"); jsonObject.put(ENDPOINT, "/ngsi-ld/v1/consumer/audit"); return jsonObject; @@ -105,7 +132,6 @@ void readFromInvalidTimeInterval(VertxTestContext testContext) { } - @Test @DisplayName("Testing read query for missing userId") void readForMissingUserId(VertxTestContext vertxTestContext) { @@ -191,7 +217,7 @@ void readForGivenTime(VertxTestContext vertxTestContext) { response -> vertxTestContext.verify( () -> { LOGGER.debug("RESPONSE" + response); - assertTrue(response.getString(TITLE).equals(SUCCESS)); + assertEquals(SUCCESS, response.getString(TITLE)); vertxTestContext.completeNow(); }))); } @@ -208,7 +234,7 @@ void readForGivenTimeAndId(VertxTestContext vertxTestContext) { response -> vertxTestContext.verify( () -> { LOGGER.debug("RESPONSE" + response); - assertTrue(response.getString(TITLE).equals(SUCCESS)); + assertEquals(SUCCESS, response.getString(TITLE)); vertxTestContext.completeNow(); }))); } @@ -225,7 +251,7 @@ void readForGivenTimeAndApi(VertxTestContext vertxTestContext) { response -> vertxTestContext.verify( () -> { LOGGER.debug("RESPONSE" + response); - assertTrue(response.getString(TITLE).equals(SUCCESS)); + assertEquals(SUCCESS, response.getString(TITLE)); vertxTestContext.completeNow(); }))); } @@ -241,7 +267,7 @@ void readForGivenTimeApiAndID(VertxTestContext vertxTestContext) { response -> vertxTestContext.verify( () -> { LOGGER.debug("RESPONSE" + response); - assertTrue(response.getString(TITLE).equals(SUCCESS)); + assertEquals(SUCCESS, response.getString(TITLE)); vertxTestContext.completeNow(); }))); } @@ -261,7 +287,8 @@ void readForGivenTimeApiAndIDEmptyResponse(VertxTestContext vertxTestContext) { () -> { LOGGER.debug( "RESPONSE " + new JsonObject(response.getMessage()).getString(DETAIL)); - assertEquals(SUCCESS, new JsonObject(response.getMessage()).getString(TITLE)); + assertEquals(RESPONSE_LIMIT_EXCEED, + new JsonObject(response.getMessage()).getString(DETAIL)); vertxTestContext.completeNow(); }))); } @@ -276,12 +303,10 @@ void countForGivenTimeApiAndIDEmptyResponse(VertxTestContext vertxTestContext) { meteringService.executeReadQuery( jsonObject, - vertxTestContext.failing( + vertxTestContext.succeeding( response -> vertxTestContext.verify( () -> { - LOGGER.debug( - "RESPONSE " + new JsonObject(response.getMessage()).getString(DETAIL)); - assertEquals(SUCCESS, new JsonObject(response.getMessage()).getString(TITLE)); + assertEquals(SUCCESS, response.getString(TITLE)); vertxTestContext.completeNow(); }))); } @@ -293,14 +318,14 @@ void writeData(VertxTestContext vertxTestContext) { request.put(USER_ID, "15c7506f-c800-48d6-adeb-0542b03947c6"); request.put(ID, "15c7506f-c800-48d6-adeb-0542b03947c6/integration-test-alias/"); request.put(API, "/ngsi-ld/v1/subscription"); - request.put(RESPONSE_SIZE,12); + request.put(RESPONSE_SIZE, 12); meteringService.executeWriteQuery( request, vertxTestContext.succeeding( response -> vertxTestContext.verify( () -> { LOGGER.debug("RESPONSE" + response.getString("title")); - assertTrue(response.getString("title").equals("Success")); + assertEquals(SUCCESS, response.getString("title")); vertxTestContext.completeNow(); }))); } @@ -356,12 +381,11 @@ void countForGivenTimeApiIdConsumerProviderID(VertxTestContext vertxTestContext) response -> vertxTestContext.verify( () -> { LOGGER.debug("RESPONSE" + response); - assertTrue(response.getString(TITLE).equals(SUCCESS)); + assertEquals(SUCCESS, response.getString(TITLE)); vertxTestContext.completeNow(); }))); } - @Disabled @Test @DisplayName("Testing count query for given time,api and providerId.") void readForGivenTimeApiAndProviderID(VertxTestContext vertxTestContext) { @@ -375,9 +399,33 @@ void readForGivenTimeApiAndProviderID(VertxTestContext vertxTestContext) { response -> vertxTestContext.verify( () -> { LOGGER.debug("RESPONSE" + response); - assertTrue(response.getString(TITLE).equals(SUCCESS)); + assertEquals(SUCCESS, response.getString(TITLE)); vertxTestContext.completeNow(); }))); } + @Test + @DisplayName("Testing read query for given time,api and resourceId where count > 10000") + void invalidReadForGivenTimeApiAndID(VertxTestContext vertxTestContext) { + JsonObject jsonObject = readConsumerRequest(); + jsonObject.put(START_TIME, "2022-05-01T14:20:00+05:30[Asia/Kolkata]"); + jsonObject.put(END_TIME, "2022-05-15T14:19:00+05:30[Asia/Kolkata]"); + jsonObject.put(API, "/ngsi-ld/v1/entities"); + jsonObject.put(RESOURCE_ID, + "iisc.ac.in/89a36273d77dac4cf38114fca1bbe64392547f86/rs.iudx.io/surat-itms-realtime-information/surat-itms-live-eta"); + + meteringService.executeReadQuery( + jsonObject, + vertxTestContext.failing( + response -> + vertxTestContext.verify( + () -> { + LOGGER.debug("RESPONSE " + response); + assertEquals( + RESPONSE_LIMIT_EXCEED, + new JsonObject(response.getMessage()).getString(DETAIL)); + vertxTestContext.completeNow(); + }))); + + } } diff --git a/src/test/java/iudx/rs/proxy/metering/util/QueryBuilderTest.java b/src/test/java/iudx/rs/proxy/metering/util/QueryBuilderTest.java deleted file mode 100644 index 8a5d3de..0000000 --- a/src/test/java/iudx/rs/proxy/metering/util/QueryBuilderTest.java +++ /dev/null @@ -1,12 +0,0 @@ -package iudx.rs.proxy.metering.util; - -import io.vertx.junit5.VertxExtension; -import org.junit.jupiter.api.extension.ExtendWith; - -import static org.junit.jupiter.api.Assertions.*; - -@ExtendWith(VertxExtension.class) -class QueryBuilderTest { - - -} \ No newline at end of file diff --git a/src/test/java/iudx/rs/proxy/metering/util/ResponseBuilderTest.java b/src/test/java/iudx/rs/proxy/metering/util/ResponseBuilderTest.java deleted file mode 100644 index e98bf8b..0000000 --- a/src/test/java/iudx/rs/proxy/metering/util/ResponseBuilderTest.java +++ /dev/null @@ -1,30 +0,0 @@ -package iudx.rs.proxy.metering.util; - -import io.vertx.junit5.VertxExtension; -import io.vertx.junit5.VertxTestContext; -import iudx.rs.proxy.common.ResponseUrn; -import org.junit.jupiter.api.DisplayName; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; - -import static iudx.rs.proxy.common.ResponseUrn.SUCCESS_URN; -import static iudx.rs.proxy.metering.util.Constants.SUCCESS; -import static org.junit.jupiter.api.Assertions.*; - -@ExtendWith(VertxExtension.class) -class ResponseBuilderTest { - - @Test - @DisplayName("Set Type And Title Test") - public void setTypeAndTitleTest(VertxTestContext vertxTestContext){ - ResponseBuilder responseBuilder= new ResponseBuilder("200"); - responseBuilder.setTypeAndTitle(200); - assertEquals("successful operations", SUCCESS_URN.getMessage()); - responseBuilder.setTypeAndTitle(204); - assertEquals("successful operations",SUCCESS_URN.getMessage()); - responseBuilder.setTypeAndTitle(400); - assertEquals("bad request parameter",ResponseUrn.BAD_REQUEST_URN.getMessage()); - vertxTestContext.completeNow(); - } - -} \ No newline at end of file