Feature/cldsrv 775 support service user

constants.js

@@ -250,6 +250,9 @@ const constants = {
     // if requester is not bucket owner, bucket policy actions should be denied with
     // MethodNotAllowed error
     onlyOwnerAllowed: ['bucketDeletePolicy', 'bucketGetPolicy', 'bucketPutPolicy'],
+    // Default value for rate limiting config cache TTL
+    rateLimitDefaultConfigCacheTTL: 30000,
+    rateLimitDefaultBurstCapacity: 1,
 };
 
 module.exports = constants;

lib/Config.js

@@ -8,7 +8,7 @@ const crypto = require('crypto');
 const { v4: uuidv4 } = require('uuid');
 const cronParser = require('cron-parser');
 const joi = require('@hapi/joi');
-const { s3routes, auth: arsenalAuth, s3middleware } = require('arsenal');
+const { s3routes, auth: arsenalAuth, s3middleware, errors, ArsenalError } = require('arsenal');
 const { isValidBucketName } = s3routes.routesUtils;
 const validateAuthConfig = arsenalAuth.inMemory.validateAuthConfig;
 const { buildAuthDataAccount } = require('./auth/in_memory/builder');
@@ -32,6 +32,7 @@ const {
     isValidType,
     isValidProtocol,
 } = require('arsenal/build/lib/network/KMSInterface');
+const { parseRateLimitConfig } = require('./api/apiUtils/rateLimit/config');
 
 // config paths
 const configSearchPaths = [
@@ -1803,6 +1804,91 @@ class Config extends EventEmitter {
         if (config.enableVeeamRoute !== undefined && config.enableVeeamRoute !== null) {
             this.enableVeeamRoute = config.enableVeeamRoute;
         }
+
+        this.rateLimiting = {
+            enabled: false,
+            bucket: {
+                configCacheTTL: constants.rateLimitDefaultConfigCacheTTL,
+            },
+        };
+
+        // {
+        //     "enabled": true,
+        //     "bucket": {
+        //         "defaultConfig": {
+        //             "requestsPerSecond": {
+        //                 "limit": 1000,
+        //                 "burstCapacity": 2
+        //             }
+        //         },
+        //         "configCacheTTL": 30000,
+        //     },
+        //     "account": {  // not in first iteration
+        //             "requestsPerSecond": {
+        //                 "limit": 1000,
+        //                 "burstCapacity": 2
+        //             }
+        //         },
+        //         "configCacheTTL": 30000,
+        //     },
+        //     "error": {
+        //         "code": 429,
+        //         "description": "Too Many Requests"
+        //     }
+        // }
+
+
+        if (config.rateLimiting?.enabled) {
+            // rate limiting uses the same localCache config defined for S3 to avoid
+            // config duplication.
+            assert(this.localCache, 'missing required property of rateLimit ' +
+                'configuration: localCache');
+
+            this.rateLimiting.enabled = true;
+
+            assert.strictEqual(typeof config.rateLimiting.serviceUserArn, 'string');
+            this.rateLimiting.serviceUserArn = config.rateLimiting.serviceUserArn;
+
+            // this.rateLimiting.default = {};
+            if (config.rateLimiting.bucket) {
+                assert.strictEqual(
+                    typeof config.rateLimiting.bucket, 'object',
+                    'rateLimiting.bucket must be an object'
+                );
+
+                let defaultConfig = undefined;
+                if (config.rateLimiting.bucket.defaultConfig) {
+                    assert.strictEqual(
+                        typeof config.rateLimiting.bucket.defaultConfig, 'object',
+                        'rateLimiting.bucket.defaultConfig must be an object'
+                    );
+                    defaultConfig = parseRateLimitConfig(config.rateLimiting.bucket.defaultConfig);
+                }
+
+                let configCacheTTL = constants.rateLimitDefaultConfigCacheTTL;
+                if (config.rateLimiting.bucket.configCacheTTL) {
+                    configCacheTTL = config.rateLimiting.bucket.configCacheTTL;
+                    assert(
+                        typeof configCacheTTL === 'number' && Number.isInteger(configCacheTTL) && configCacheTTL > 0,
+                        'ratelimiting.bucket.configCacheTTL must be a postive integer'
+                    );
+                }
+
+                this.rateLimiting.bucket = {
+                    defaultConfig,
+                    configCacheTTL,
+                };
+            }
+
+            if (config.error) {
+                assert.strictEqual(typeof config.rateLimiting.error, 'object', 'rateLimiting.error must be an object');
+                this.rateLimiting.error = new ArsenalError('SlowDown',
+                    config.throttling.errorResponse.code, config.throttling.errorResponse.message);
+            } else {
+                this.rateLimiting.error = errors.SlowDown;
+            }
+        }
+
         return config;
     }
 

lib/api/apiUtils/authorization/serviceUser.js

@@ -0,0 +1,39 @@
+const { timingSafeEqual } = require('crypto');
+
+const { config } = require('../../../Config');
+
+function isRateLimitServiceUser(authInfo, log) {
+    try {
+        const requestArn = authInfo.getArn();
+        const configuredArn = config.rateLimiting?.serviceUserArn;
+
+        if (!configuredArn) {
+            log.warn('Access denied - no configured ARN under rateLimiting.serviceUserArn', { requestArn });
+            return false;
+        }
+
+        // Support partial ARN matching (e.g., match by account only)
+        // If configuredArn is shorter, do prefix match; otherwise exact match
+        let match;
+        if (requestArn.length >= configuredArn.length) {
+            // Extract prefix from requestArn to match configuredArn length
+            const requestPrefix = requestArn.substring(0, configuredArn.length);
+            match = timingSafeEqual(
+                Buffer.from(requestPrefix),
+                Buffer.from(configuredArn)
+            );
+        } else {
+            // requestArn is shorter than configured - no match
+            log.warn('Access denied - request ARN is shorter than configured ARN', { requestArn, configuredArn });
+            match = false;
+        }
+        return match;
+    } catch (err) {
+        log.error('Error checking if request is rate limit service user', { error: err });
+        return false;
+    }
+}
+
+module.exports = {
+    isRateLimitServiceUser
+};

lib/api/apiUtils/rateLimit/cache.js

@@ -0,0 +1,74 @@
+const counters = new Map();
+
+const configCache = new Map();
+
+function setCounter(key, value) {
+    // Make sure that the Map remains in order
+    // Counters expiring soonest will be first during iteration.
+    counters.delete(key);
+    counters.set(key, value);
+}
+
+function getCounter(key) {
+    return counters.get(key);
+}
+
+function expireCounters(now) {
+    const toRemove = [];
+    for (const [key, value] of counters.entries()) {
+        if (value <= now) {
+            toRemove.push(key);
+        }
+    }
+
+    for (const key of toRemove) {
+        counters.delete(key);
+    }
+}
+
+function setCachedConfig(key, limitConfig, ttl) {
+    const expiry = Date.now() + ttl;
+    configCache.set(key, { expiry, config: limitConfig });
+}
+
+function getCachedConfig(key) {
+    const value = configCache.get(key);
+    if (value === undefined) {
+        return undefined;
+    }
+
+    const { expiry, config } = value;
+    if (expiry <= Date.now()) {
+        configCache.delete(key);
+        return undefined;
+    }
+
+    return config;
+}
+
+function expireCachedConfigs(now) {
+    const toRemove = [];
+    for (const [key, { expiry }] of configCache.entries()) {
+        if (expiry <= now) {
+            toRemove.push(key);
+        }
+    }
+
+    for (const key of toRemove) {
+        configCache.delete(key);
+    }
+}
+
+module.exports = {
+    setCounter,
+    getCounter,
+    expireCounters,
+    setCachedConfig,
+    getCachedConfig,
+    expireCachedConfigs,
+
+    // Do not access directly
+    // Used only for tests
+    counters,
+    configCache,
+};

lib/api/apiUtils/rateLimit/client.js

@@ -0,0 +1,78 @@
+const fs = require('fs');
+
+const Redis = require('ioredis');
+
+const { config } = require('../../../Config');
+
+const updateCounterScript = fs.readFileSync(`${__dirname  }/updateCounter.lua`).toString();
+
+const SCRIPTS = {
+    updateCounter: {
+        numberOfKeys: 1,
+        lua: updateCounterScript,
+    },
+};
+
+class RateLimitClient {
+    constructor(redisConfig) {
+        this.redis = new Redis({
+            ...redisConfig,
+            scripts: SCRIPTS,
+            lazyConnect: true,
+        });
+    }
+
+    /**
+     * @typedef {Object} CounterUpdateBatch
+     * @property {string} key - counter key
+     * @property {number} cost - cost to add to counter
+     */
+
+    /**
+     * @typedef {Object} CounterUpdateBatchResult
+     * @property {string} key - counter key
+     * @property {number} value - current value of counter
+     */
+
+    /**
+     * @callback RateLimitClient~batchUpdate
+     * @param {Error|null} err
+     * @param {CounterUpdateBatchResult[]|undefined}
+     */
+
+    /**
+     * Add cost to the counter at key.
+     * Returns the new value for  the counter
+     *
+     * @param {CounterUpdateBatch[]} batch - batch of counter updates
+     * @param {RateLimitClient~batchUpdate} cb
+     */
+    updateLocalCounters(batch, cb) {
+        const pipeline = this.redis.pipeline();
+        for (const { key, cost } of batch) {
+            pipeline.updateCounter(key, cost);
+        }
+
+        pipeline.exec((err, results) => {
+            if (err) {
+                cb(err);
+                return;
+            }
+
+            cb(null, results.map((res, i) => ({
+                key: batch[i].key,
+                value: res[1],
+            })));
+        });
+    }
+}
+
+let instance;
+if (config.rateLimiting.enabled) {
+    instance = new RateLimitClient(config.localCache);
+}
+
+module.exports = {
+    instance,
+    RateLimitClient
+};

lib/api/apiUtils/rateLimit/config.js

@@ -0,0 +1,32 @@
+const assert = require('assert');
+
+const { rateLimitDefaultBurstCapacity } = require('../../../../constants');
+
+function parseRateLimitConfig(config) {
+    const limitConfig = {};
+
+    if (config.requestsPerSecond) {
+        assert.strictEqual(typeof config.requestsPerSecond, 'object');
+
+        const { limit } = config.requestsPerSecond;
+        assert(typeof limit === 'number' && Number.isInteger(limit) && limit > 0);
+
+        let { burstCapacity } = config.requestsPerSecond;
+        if (burstCapacity !== undefined) {
+            assert(typeof burstCapacity === 'number' && Number.isInteger(burstCapacity) && burstCapacity > 0);
+        } else {
+            burstCapacity = rateLimitDefaultBurstCapacity;
+        }
+
+        limitConfig.requestsPerSecond = {
+            interval: 1000 / limit,
+            bucketSize: burstCapacity * 1000,
+        };
+    }
+
+    return limitConfig;
+}
+
+module.exports = {
+    parseRateLimitConfig,
+};

lib/api/apiUtils/rateLimit/updateCounter.lua

@@ -0,0 +1,27 @@
+-- updateCounter <KEY> <COST>
+--
+-- Adds the passed COST to the GCRA counter at KEY.
+-- If no counter currently exists a new one is created from the current time.
+-- The key expiration is set to the updated value.
+-- Returns the value of the updated key.
+
+local ts = redis.call('TIME')
+local currentTime = ts[1] * 1000
+currentTime = currentTime + math.floor(ts[2] / 1000)
+
+local newValue = currentTime + tonumber(ARGV[1])
+
+local counterExists = redis.call('EXISTS', KEYS[1])
+if counterExists == 1 then
+    local currentValue = tonumber(redis.call('GET', KEYS[1]))
+    if currentValue > currentTime then
+        newValue = currentValue + tonumber(ARGV[1])
+    end
+end
+
+redis.call('SET', KEYS[1], newValue)
+
+local expiry = math.ceil(newValue / 1000)
+redis.call('EXPIREAT', KEYS[1], expiry)
+
+return newValue