More cache namespaces for documentation: glossary etc. IP rate limiting penalty

.github/workflows/run_trivy.yml

@@ -20,7 +20,9 @@ env:
 jobs:
   build:
     runs-on: ubuntu-latest
-    if: ${{ github.event_name == 'workflow_dispatch' || github.event.workflow_run.conclusion == 'success' }}
+    # Only scan when build_container.yml actually pushed an image: its docker job
+    # is gated on the same variable, so without it there is nothing to pull.
+    if: ${{ vars.ENABLE_CONTAINER_BUILDING == 'true' && (github.event_name == 'workflow_dispatch' || github.event.workflow_run.conclusion == 'success') }}
 
     steps:
       - uses: actions/checkout@v4
@@ -40,8 +42,9 @@ jobs:
           path: .trivy
           key: ${{ runner.os }}-trivy-db-${{ steps.trivy-db.outputs.sha }}
       - name: Run Trivy vulnerability scanner
-        uses: aquasecurity/trivy-action@ed142fd0673e97e23eac54620cfb913e5ce36c25
+        uses: aquasecurity/trivy-action@ed142fd0673e97e23eac54620cfb913e5ce36c25 # v0.36.0
         with:
+          version: "v0.74.0"
           image-ref: "docker.io/${{ env.DOCKER_HUB_ORGANIZATION }}/${{ env.DOCKER_HUB_REPOSITORY }}:${{ github.event_name == 'workflow_dispatch' && (inputs.image_sha || github.sha) || github.event.workflow_run.head_sha }}"
           format: "template"
           template: "@/contrib/sarif.tpl"

docs/telemetry_conventions.md

@@ -288,6 +288,8 @@ thread and class figures on its own port. Remove it once the Micrometer JVM bind
 | `obp.api.batch_writer.queue.depth` | gauge | `writer` | rows queued minus rows written or lost (the queue's own `size()` walks it) |
 | `obp.api.batch_writer.flushes` | timer | `writer`, `result` | each flush that had rows |
 | `obp.api.redis_cache.gets`, `obp.api.redis_cache.sets` | counter | `cache` (`static_resource_docs`, `dynamic_resource_docs`, `all_resource_docs`, `static_swagger`, `financial_products`, `api_products`), `result` (`hit`, `miss`, `error` for gets; `success`, `error` for sets) | `Caching.tryGet` / `trySet`; `error` means Redis was unreachable |
+| `obp.api.ip_penalties.active` | gauge | | how many addresses have an IP penalty (never which ones) |
+| `obp.api.ip_penalties.refused` | counter | | requests refused with 429 OBP-10062 |
 | `obp.api.self_service_rate_limit.checks` | counter | `scope`, `outcome` (`allowed`, `warned`, `blocked`, `skipped`) | `SelfServiceRateLimiter.check`; in shadow mode, `warned` shows how often a scope would refuse real traffic |
 | `obp.api.connector.calls` | timer, fixed buckets | `connector`, `connector_method`, `result` | the Connector proxy (`code/bankconnectors/package.scala`) |
 | `obp.api.redis.commands` | timer | `command`, `result` | `Redis.use` |

obp-api/src/main/scala/code/api/cache/Caching.scala

@@ -143,13 +143,19 @@ object Caching extends MdcLoggable {
   def setAllResourceDocCache(key: String, value: String): Unit =
     trySet("all_resource_docs", ALL_RESOURCE_DOC_CACHE_KEY_PREFIX, key, GET_DYNAMIC_RESOURCE_DOCS_TTL, value)
 
-  // Also holds the connector JSON Schemas served by v6.0.0 message-docs/CONNECTOR/json-schema.
   def getStaticSwaggerDocCache(key: String): Option[String] =
     tryGet("static_swagger", STATIC_SWAGGER_DOC_CACHE_KEY_PREFIX, key, GET_STATIC_RESOURCE_DOCS_TTL)
 
   def setStaticSwaggerDocCache(key: String, value: String): Unit =
     trySet("static_swagger", STATIC_SWAGGER_DOC_CACHE_KEY_PREFIX, key, GET_STATIC_RESOURCE_DOCS_TTL, value)
 
+  // The rendered message docs and the JSON Schema of each connector (namespace message_docs).
+  def getMessageDocsCache(key: String): Option[String] =
+    tryGet("message_docs", MESSAGE_DOCS_CACHE_KEY_PREFIX, key, GET_STATIC_RESOURCE_DOCS_TTL)
+
+  def setMessageDocsCache(key: String, value: String): Unit =
+    trySet("message_docs", MESSAGE_DOCS_CACHE_KEY_PREFIX, key, GET_STATIC_RESOURCE_DOCS_TTL, value)
+
   // Fail-safe wrappers around Redis.use. If Redis is unreachable (dev without a
   // running Redis, transient failure, etc.) we treat it as a miss and recompute instead of failing
   // the whole request.

obp-api/src/main/scala/code/api/constant/constant.scala

@@ -164,6 +164,32 @@
     }
   }
 
+  /** How long [[recentCacheNamespaceVersion]] trusts its copy of a namespace version before re-reading Redis. */
+  final val RecentNamespaceVersionMillis = 1000L
+
+  private val recentNamespaceVersions = new java.util.concurrent.ConcurrentHashMap[String, (Long, Long)]()
+
+  /**
+   * The version of a cache namespace, re-read from Redis at most once every
+   * [[RecentNamespaceVersionMillis]]. For caches held in each instance's memory (message docs,
+   * JSON Schemas, the Glossary), which put the version in their own keys so that bumping the
+   * namespace reaches them too, without a Redis read on every request. After a bump, the instance
+   * that made it sees the new version at once and the others within a second.
+   */
+  def recentCacheNamespaceVersion(namespaceId: String): Long = {
+    val now = System.currentTimeMillis()
+    Option(recentNamespaceVersions.get(namespaceId)) match {
+      case Some((version, readAt)) if now - readAt < RecentNamespaceVersionMillis => version
+      case _ =>
+        val version = getCacheNamespaceVersion(namespaceId)
+        recentNamespaceVersions.put(namespaceId, (version, now))
+        version
+    }
+  }
+
+  /** Forget the local copies of namespace versions, so the next read goes to Redis (tests). */
+  def forgetRecentCacheNamespaceVersions(): Unit = recentNamespaceVersions.clear()
+
   /**
    * Increment the version counter for a cache namespace.
    * This effectively invalidates all cached keys in that namespace by making them unreachable.
@@ -182,6 +208,8 @@
       val newVersion = Redis.use(JedisMethod.INCR, versionKey, None, None)
         .map(_.toLong)
       logger.info(s"Cache namespace version incremented: ${namespaceId} -> ${newVersion.getOrElse("unknown")}")
+      // This instance sees the new version at once; the others within RecentNamespaceVersionMillis.
+      newVersion.foreach(v => recentNamespaceVersions.put(namespaceId, (v, System.currentTimeMillis())))
       newVersion
     } catch {
       case e: Throwable =>
@@ -340,6 +368,12 @@
   final val CONNECTOR_INBOUND_NAMESPACE = "connector_inbound"
   final val FINANCIAL_PRODUCTS_NAMESPACE = "financial_products"
   final val API_PRODUCTS_NAMESPACE = "api_products"
+  // The rendered message docs of each connector (GET /message-docs/CONNECTOR) and each connector's
+  // JSON Schema (GET /message-docs/CONNECTOR/json-schema), in Redis and in each instance's memory.
+  final val MESSAGE_DOCS_NAMESPACE = "message_docs"
+  // The Glossary as each instance holds it in memory. Bumping it reloads the Glossary at once and
+  // rebuilds every cached resource-docs document, because their keys carry the Glossary version.
+  final val GLOSSARY_NAMESPACE = "glossary"
 
   // List of all versioned cache namespaces
   final val ALL_CACHE_NAMESPACES = List(
@@ -357,7 +391,9 @@
     CONNECTOR_OUTBOUND_NAMESPACE,
     CONNECTOR_INBOUND_NAMESPACE,
     FINANCIAL_PRODUCTS_NAMESPACE,
-    API_PRODUCTS_NAMESPACE
+    API_PRODUCTS_NAMESPACE,
+    MESSAGE_DOCS_NAMESPACE,
+    GLOSSARY_NAMESPACE
   )
 
   // Cache key prefixes with global namespace and versioning for easy invalidation
@@ -368,6 +404,7 @@
   def STATIC_RESOURCE_DOC_CACHE_KEY_PREFIX: String = getVersionedCachePrefix(RD_STATIC_NAMESPACE)
   def ALL_RESOURCE_DOC_CACHE_KEY_PREFIX: String = getVersionedCachePrefix(RD_ALL_NAMESPACE)
   def STATIC_SWAGGER_DOC_CACHE_KEY_PREFIX: String = getVersionedCachePrefix(SWAGGER_STATIC_NAMESPACE)
+  def MESSAGE_DOCS_CACHE_KEY_PREFIX: String = getVersionedCachePrefix(MESSAGE_DOCS_NAMESPACE)
   final val CREATE_LOCALISED_RESOURCE_DOC_JSON_TTL: Int = APIUtil.getPropsValue(s"createLocalisedResourceDocJson.cache.ttl.seconds", "3600").toInt
   final val GET_DYNAMIC_RESOURCE_DOCS_TTL: Int = APIUtil.getPropsValue(s"dynamicResourceDocsObp.cache.ttl.seconds", "3600").toInt
   final val GET_STATIC_RESOURCE_DOCS_TTL: Int = APIUtil.getPropsValue(s"staticResourceDocsObp.cache.ttl.seconds", "3600").toInt

obp-api/src/main/scala/code/api/util/ApiRole.scala

@@ -569,6 +569,17 @@ object ApiRole extends MdcLoggable{
   case class CanGetTelemetry(requiresBankId: Boolean = false) extends ApiRole
   lazy val canGetTelemetry = CanGetTelemetry()
 
+  // IP penalties restrict an address on the whole instance, which belongs to no bank, so these
+  // Roles are held at the empty bank id.
+  case class CanCreateIpPenalty(requiresBankId: Boolean = false) extends ApiRole
+  lazy val canCreateIpPenalty = CanCreateIpPenalty()
+
+  case class CanGetIpPenalties(requiresBankId: Boolean = false) extends ApiRole
+  lazy val canGetIpPenalties = CanGetIpPenalties()
+
+  case class CanDeleteIpPenalty(requiresBankId: Boolean = false) extends ApiRole
+  lazy val canDeleteIpPenalty = CanDeleteIpPenalty()
+
   case class CanGetSignalStats(requiresBankId: Boolean = false) extends ApiRole
   lazy val canGetSignalStats = CanGetSignalStats()
 

obp-api/src/main/scala/code/api/util/ErrorMessages.scala

@@ -138,6 +138,12 @@ object ErrorMessages {
   //  OBP-10061 the authentication limiter (AuthRateLimiter, inside the credential check, keyed by IP and account)
   val TooManyRequestsSelfService = "OBP-10060: Too Many Requests for a self-service endpoint."
   val TooManyRequestsAuth = "OBP-10061: Too Many Requests for authentication. Too many login attempts from this address or for this account."
+  //  OBP-10062 an IP penalty (IpPenalties, before everything else, an operator's temporary limit on one address)
+  val TooManyRequestsIpPenalty = "OBP-10062: Too Many Requests. This address is under a temporary rate limit set by an operator."
+  val InvalidIpAddress = "OBP-10063: Invalid IP address. Give an IPv4 or IPv6 address, not a host name."
+  val IpPenaltyAlreadyExists = "OBP-10064: This IP address already has a penalty. Remove it first to change it."
+  val IpPenaltyNotFound = "OBP-10065: This IP address has no penalty."
+  val InvalidIpPenalty = "OBP-10066: Invalid IP penalty. per_minute_limit must be 0 or more, duration_minutes between 1 and 10080 (one week), and reason between 1 and 255 characters."
   // Not an error: the text of the X-Rate-Limit-Warning header a self-service endpoint returns in
   // shadow mode. SCOPE and LIMIT are replaced at runtime, e.g. "signup" and "5 per hour".
   // See SelfServiceRateLimiter.warningMessage.

obp-api/src/main/scala/code/api/util/Glossary.scala

@@ -168,7 +168,11 @@ object Glossary extends MdcLoggable  {
 		val (checkedAt, version, byTitle) = cachedItemsByTitle.get()
 		if (checkedAt != 0L && now - checkedAt < GlossaryCacheRecheckMillis) (version, byTitle)
 		else {
-			val currentVersion = dynamicGlossaryItemsVersion
+			// The glossary cache namespace version is part of the token: bumping it (for example from
+			// the cache page in API Manager) reloads the Glossary here and, because the token is in
+			// every resource-docs cache key, rebuilds every cached document that embeds Glossary text.
+			val currentVersion =
+				s"$dynamicGlossaryItemsVersion-ns${code.api.Constant.recentCacheNamespaceVersion(code.api.Constant.GLOSSARY_NAMESPACE)}"
 			if (checkedAt != 0L && currentVersion == version) {
 				cachedItemsByTitle.set((now, version, byTitle))
 				(version, byTitle)
@@ -188,6 +192,9 @@ object Glossary extends MdcLoggable  {
 
 	private def glossaryItemsByTitle: Map[String, GlossaryItem] = glossaryState._2
 
+	/** Glossary items this instance holds in memory now (static and dynamic), for the cache page. */
+	def loadedItemCount: Int = glossaryItemsByTitle.size
+
 	/**
 	 * A token for Resource Doc cache keys. It changes whenever a Dynamic Glossary Item is added,
 	 * changed or removed, so a cached endpoint description that embeds Glossary text is rebuilt
@@ -763,6 +770,8 @@ object Glossary extends MdcLoggable  {
 				 |2. **Authentication limiter** (`auth.rate_limit.*`) runs inside the credential check of Direct Login, DAuth, Gateway Login and SIWE, before the password or token is verified, keyed by IP address and by account. It defends against brute force, credential stuffing and lockout attacks. Trip code: `OBP-10061`.
 				 |3. **Consumer quota** (the limits described above) runs after authentication, keyed by Consumer, or by IP address with a single hourly ceiling for anonymous calls. It is the commercial and fair-use quota. Trip code: `OBP-10018`.
 				 |
+				 |Before all three, an operator can put a single IP address under a temporary **IP penalty**: a per-minute limit on every endpoint, for a set time, for example during a scan or denial-of-service attempt (`POST /obp/v7.0.0/management/ip-penalties`, Role CanCreateIpPenalty). A per-minute limit of 0 refuses every request. Penalties are always enforced, shared by every instance, and disappear when they expire; the penalty endpoints themselves are never refused, so a mistake can be undone. Trip code: `OBP-10062`.
+				 |
 				 |A login attempt is counted by the authentication limiter only; it is not a self-service scope, so no attempt is counted twice. Every limiter counts in Redis and fails open: a Redis outage never blocks a call.
 				 |
 				 |### Self-service rate limiting (per IP address, before any credential)