apps/spring/token-exchange-lib/src/main/java/com/tokenexchange/lib/RevocationCache.java - master - Al/TESamples

apps/spring/token-exchange-lib/src/main/java/com/tokenexchange/lib/RevocationCache.java
159 строк6 KB
Initial Commit
11 май 2026, 17:40
11 май 2026, 17:400a04577
package com.tokenexchange.lib;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

import java.util.concurrent.atomic.AtomicLong;

/**
 * In-memory список запрета (по UUID пользователя), наполняемый через
 * Redis pub/sub (канал {@code kc-events}). Реализует §6.5 Рекомендаций
 * для клиента «Выход пользователя через обратный канал»
 * (docs/client-recommendations.md#65) на стороне chain-сервисов.
 *
 * <p>Источник событий — обработчик в Keycloak, который слушает {@code LOGOUT},
 * {@code REVOKE_GRANT}, {@code USER_LOGGED_OUT_BY_ADMIN} и публикует JSON
 * в канал. Подписка — daemon-поток с Jedis.
 *
 * <p>Использование (Spring chain-service):
 * <pre>
 *   var cache = new RevocationCache("redis", 6379, "kc-events");
 *   cache.start();
 *   // в EchoController:
 *   if (cache.isRevoked(jwt.getSubject())) throw new ResponseStatusException(401, "session_revoked");
 * </pre>
 *
 * Что отложено:
 *  - TTL для записей deny-list (по token.exp - now). Сейчас множество растёт
 *    бесконечно — для demo это не критично, но в production обязательно.
 *  - Distributed L2 (Phase 8 / FR-04) — каждая реплика держит свой in-memory
 *    Set; pub/sub доносит инвалидацию во все одновременно.
 */
public class RevocationCache implements AutoCloseable {

    private static final Logger log = LoggerFactory.getLogger(RevocationCache.class);
    private static final ObjectMapper M = new ObjectMapper();

    private final String host;
    private final int port;
    private final String channel;
    /** Phase 12: TTL для deny-list записи. Хранится {sub: expirationEpochMillis}. */
    private final java.util.concurrent.ConcurrentHashMap<String, Long> revokedSubs =
        new java.util.concurrent.ConcurrentHashMap<>();
    private final long defaultTtlMillis;
    private final AtomicLong eventsReceived = new AtomicLong();
    private volatile Thread subscriberThread;
    private volatile Thread cleanupThread;
    private volatile boolean closed;

    public RevocationCache(String host, int port, String channel) {
        this(host, port, channel, java.time.Duration.ofMinutes(60).toMillis());
    }

    public RevocationCache(String host, int port, String channel, long defaultTtlMillis) {
        this.host = host;
        this.port = port;
        this.channel = channel;
        this.defaultTtlMillis = defaultTtlMillis;
    }

    public void start() {
        if (subscriberThread != null) return;
        subscriberThread = new Thread(this::subscribeLoop, "revocation-subscriber");
        subscriberThread.setDaemon(true);
        subscriberThread.start();
        // Phase 12 / FR-05: фоновый cleanup истёкших deny-list записей.
        cleanupThread = new Thread(this::cleanupLoop, "revocation-cleanup");
        cleanupThread.setDaemon(true);
        cleanupThread.start();
        log.info("RevocationCache: subscribing to redis://{}:{}/{}, default TTL = {} ms",
            host, port, channel, defaultTtlMillis);
    }

    private void cleanupLoop() {
        while (!closed) {
            try {
                Thread.sleep(30_000);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
                return;
            }
            long now = System.currentTimeMillis();
            int removed = 0;
            var it = revokedSubs.entrySet().iterator();
            while (it.hasNext()) {
                var e = it.next();
                if (e.getValue() < now) { it.remove(); removed++; }
            }
            if (removed > 0) log.debug("RevocationCache: cleaned {} expired entries", removed);
        }
    }

    private void subscribeLoop() {
        while (!closed) {
            try (Jedis jedis = new Jedis(host, port)) {
                jedis.subscribe(new JedisPubSub() {
                    @Override
                    public void onMessage(String ch, String msg) {
                        eventsReceived.incrementAndGet();
                        try {
                            JsonNode n = M.readTree(msg);
                            String user = n.path("user").asText("");
                            String type = n.path("type").asText("");
                            if (!user.isEmpty()) {
                                long expiration = System.currentTimeMillis() + defaultTtlMillis;
                                revokedSubs.put(user, expiration);
                                log.info("revoked sub={} (event={}, ttl_ms={})",
                                    user, type, defaultTtlMillis);
                            } else {
                                log.debug("event without user (skipped): {}", msg);
                            }
                        } catch (Exception e) {
                            log.warn("Failed to parse event '{}': {}", msg, e.getMessage());
                        }
                    }
                }, channel);
            } catch (Exception e) {
                if (closed) return;
                log.warn("Redis subscriber error, reconnecting in 1s: {}", e.getMessage());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        }
    }

    public boolean isRevoked(String sub) {
        if (sub == null) return false;
        Long expiration = revokedSubs.get(sub);
        if (expiration == null) return false;
        if (expiration < System.currentTimeMillis()) {
            // Lazy expiry: запись истекла, удаляем.
            revokedSubs.remove(sub, expiration);
            return false;
        }
        return true;
    }

    public long eventsReceived() {
        return eventsReceived.get();
    }

    public int revokedCount() {
        return revokedSubs.size();
    }

    @Override
    public void close() {
        closed = true;
        if (subscriberThread != null) subscriberThread.interrupt();
        if (cleanupThread != null) cleanupThread.interrupt();
    }
}