From d6eb25a0f6a5d12770abbfbff9a22958832662c5 Mon Sep 17 00:00:00 2001 From: root Date: Wed, 21 Oct 2026 09:57:46 +0800 Subject: [PATCH] Record standby subscription events with signed durable receipts --- docs/SUBSCRIPTION_EVENTS.md | 30 ++ scripts/migrate_subscription_events.py | 80 +++++ scripts/test_subscription_events.py | 82 +++++ .../SubscriptionEventController.java | 33 ++ .../lionwebsite/Dao/normal/SubMapper.java | 2 +- .../Dao/normal/SubscriptionEventMapper.java | 31 ++ .../lionwebsite/Domain/SubscriptionEvent.java | 16 + .../Domain/SubscriptionEventBinding.java | 10 + .../lion/lionwebsite/Service/SubService.java | 17 +- .../Service/SubscriptionEventService.java | 222 ++++++++++++ .../SubscriptionStandbySnapshotService.java | 8 +- .../MyBatisNativeConfigurationTest.java | 16 + .../SubscriptionEventIntegrationTest.java | 317 ++++++++++++++++++ 13 files changed, 860 insertions(+), 4 deletions(-) create mode 100644 docs/SUBSCRIPTION_EVENTS.md create mode 100644 scripts/migrate_subscription_events.py create mode 100644 scripts/test_subscription_events.py create mode 100644 src/main/java/com/lion/lionwebsite/Controller/SubscriptionEventController.java create mode 100644 src/main/java/com/lion/lionwebsite/Dao/normal/SubscriptionEventMapper.java create mode 100644 src/main/java/com/lion/lionwebsite/Domain/SubscriptionEvent.java create mode 100644 src/main/java/com/lion/lionwebsite/Domain/SubscriptionEventBinding.java create mode 100644 src/main/java/com/lion/lionwebsite/Service/SubscriptionEventService.java create mode 100644 src/test/java/com/lion/lionwebsite/Service/SubscriptionEventIntegrationTest.java diff --git a/docs/SUBSCRIPTION_EVENTS.md b/docs/SUBSCRIPTION_EVENTS.md new file mode 100644 index 0000000..759803e --- /dev/null +++ b/docs/SUBSCRIPTION_EVENTS.md @@ -0,0 +1,30 @@ +# 备用订阅更新事件 + +主站 `/sub/{client}/{key}` 接受网关的签名事件；`GET` 在客户端类型与缓存路径有效后，同一个主库事务写更新记录和长期收据。签名有效的 `HEAD` 不写记录。普通请求以及无效签名仍走原来的更新记录逻辑。 + +`POST /internal/subscription-events` 有界读取原始请求（最多 256 KiB），批次最多 100 条。签名算法为 HMAC-SHA256，同步密钥来自 `SUBSCRIPTION_SYNC_SECRET`；签名输入是 `v1\n` + epoch 秒时间戳 + `\n` + 原始请求体 SHA256 十六进制。时间偏差最多 300 秒。请求头为 `X-Subscription-Timestamp` 和 `X-Subscription-Signature`。主站 GET 将单个事件的原始 JSON 使用无 padding 的 base64url 编码放入 `X-Subscription-Event`，同样对原始事件 JSON 签名。 + +事件字段：`eventId`（规范小写 UUID）、`keyHash`（公开 key 的小写 SHA256）、`client`（v2 或 cat）、`clientIp`（IPv4/IPv6 字面值）、`userAgent`（最多 2048 字符）、`occurredAt`（epoch 毫秒）。允许历史事件，事件时间不可超过未来 5 分钟。签名中的 IP 为权威来源，不能由未经签名的 XFF 覆盖。 + +响应 `acknowledged` 只在事务已提交后返回，包括已存在的相同事件。冲突事件 ID 返回 400；签名无效返回 403；数据库异常返回 5xx，无确认列表。未知 hash 不确认；历史映射中用户已删除则写终态收据并确认，不猜测用户归属。可见记录按事件时间和 ID 倒序保留每用户最近 10 条，收据独立长期保留。批次任意数据库写入失败会全部回滚。 + +## 显式迁移 + +运行前先执行稳定 user_id 关联迁移。新表不会在启动时创建。 + +1. 先在数据库副本上演练：`python3 scripts/migrate_subscription_events.py /path/to/copy.db`。 +2. 停止生产主站服务，确认没有其他写入者。 +3. 执行 `python3 scripts/migrate_subscription_events.py /path/to/main.db --apply --service-stopped`。 +4. 检查输出 integrity 为 ok，保留输出中的 SQLite 备份，然后启动新程序。 + +脚本在写入前使用 SQLite backup API 生成权限 0600 的备份，在单个事务中建表并初始化历史 hash，重复执行保留既有收据和历史。不将公开 key 写入历史表或输出；原数据库备份含既有敏感数据，须保留原有保护措施。 + +快照构建记住所有现有绑定；重置或删除绑定前也保存历史 hash。用户改名时回传事件按 stable user_id 读取当前名字。合法旧订阅标签的 user_id 可以为 NULL，历史映射保留标签并按原标签写回，不创建用户，也不猜测同名新用户归属。可见日志删除不删除历史映射或收据。 + +Java 编译与测试必须在部署手册指定的 192.168.5.30 执行： + +```sh +mvn -Dtest=SubscriptionEventIntegrationTest,SubServiceTest,SubscriptionStandbySnapshotServiceTest,MyBatisNativeConfigurationTest test +``` + +Python 迁移验证可本机执行：`python3 -m unittest discover -s scripts -p 'test_subscription_events.py' -v`。 diff --git a/scripts/migrate_subscription_events.py b/scripts/migrate_subscription_events.py new file mode 100644 index 0000000..64310ec --- /dev/null +++ b/scripts/migrate_subscription_events.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python3 +"""Create durable subscription event receipts/history, with a SQLite backup. + +Stop the production backend before --apply and pass --service-stopped to confirm. +No automatic startup DDL exists. Dry run is read-only; only hashes enter history. +Receipt retention is independent of the most recent ten visible update records. +""" +import argparse +import datetime +import hashlib +import json +from pathlib import Path +import sqlite3 + +DDL = ( + """CREATE TABLE IF NOT EXISTS subscription_event_receipt ( + event_id TEXT PRIMARY KEY NOT NULL, key_hash TEXT NOT NULL, + client TEXT NOT NULL, occurred_at INTEGER NOT NULL, source TEXT NOT NULL, + ingested_at INTEGER NOT NULL, fingerprint TEXT NOT NULL, outcome TEXT NOT NULL + )""", + """CREATE TABLE IF NOT EXISTS subscription_event_binding ( + key_hash TEXT PRIMARY KEY NOT NULL, user_id INTEGER, + user_name TEXT NOT NULL, first_seen_at INTEGER NOT NULL, last_seen_at INTEGER NOT NULL + )""", + "CREATE INDEX IF NOT EXISTS idx_subscription_event_binding_user ON subscription_event_binding(user_id)", + "CREATE INDEX IF NOT EXISTS idx_sub_record_user_time ON sub_update_record(user_id,time DESC,id DESC)", +) + + +def migrate(database: Path, apply=False, service_stopped=False): + if not database.is_file(): + raise ValueError("database must already exist") + if apply and not service_stopped: + raise ValueError("stop backend and supply --service-stopped before applying") + uri = database.resolve().as_uri() + ("?mode=rw" if apply else "?mode=ro") + with sqlite3.connect(uri, uri=True) as db: + for table in ("User", "sub_bind", "sub_update_record"): + columns = {row[1] for row in db.execute(f'PRAGMA table_info("{table}")')} + if not columns or (table != "User" and "user_id" not in columns): + raise ValueError("stable user links migration is required first") + unresolved = db.execute("SELECT count(*) FROM sub_bind sb LEFT JOIN User u ON u.id=sb.user_id WHERE sb.user_id IS NOT NULL AND u.id IS NULL").fetchone()[0] + if unresolved: + raise ValueError("resolve non-null dangling subscription user IDs before migration") + tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'")} + result = {"apply": apply, "tables_to_create": sorted({"subscription_event_receipt", "subscription_event_binding"} - tables)} + if not apply: + return result + stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") + backup = database.with_name(database.name + ".before-subscription-events." + stamp) + # Create mode 600 before any sensitive database bytes are written. + backup.touch(mode=0o600, exist_ok=False) + with sqlite3.connect(backup) as snapshot: + db.backup(snapshot) + try: + db.execute("BEGIN IMMEDIATE") + for statement in DDL: + db.execute(statement) + now = int(datetime.datetime.now(datetime.timezone.utc).timestamp() * 1000) + for key, user_id, name in db.execute("SELECT sb.key,sb.user_id,coalesce(u.username,sb.user) FROM sub_bind sb LEFT JOIN User u ON u.id=sb.user_id"): + key_hash = hashlib.sha256(key.encode("utf-8")).hexdigest() + db.execute("""INSERT INTO subscription_event_binding VALUES(?,?,?,?,?) + ON CONFLICT(key_hash) DO UPDATE SET user_name=excluded.user_name,last_seen_at=excluded.last_seen_at + WHERE subscription_event_binding.user_id IS excluded.user_id""", (key_hash, user_id, name, now, now)) + db.commit() + except BaseException: + db.rollback() + raise + result["backup"] = str(backup) + result["history_bindings"] = db.execute("SELECT count(*) FROM subscription_event_binding").fetchone()[0] + result["integrity"] = db.execute("PRAGMA quick_check").fetchone()[0] + return result + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("database", type=Path) + parser.add_argument("--apply", action="store_true") + parser.add_argument("--service-stopped", action="store_true") + args = parser.parse_args() + print(json.dumps(migrate(args.database, args.apply, args.service_stopped), ensure_ascii=False)) diff --git a/scripts/test_subscription_events.py b/scripts/test_subscription_events.py new file mode 100644 index 0000000..7b46b2c --- /dev/null +++ b/scripts/test_subscription_events.py @@ -0,0 +1,82 @@ +import importlib.util +from pathlib import Path +import sqlite3 +import tempfile +import unittest +import hashlib + +spec = importlib.util.spec_from_file_location("migration", Path(__file__).with_name("migrate_subscription_events.py")) +migration = importlib.util.module_from_spec(spec) +spec.loader.exec_module(migration) + + +class SubscriptionEventsMigrationTest(unittest.TestCase): + def test_explicit_backup_atomic_and_idempotent(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "main.db" + with sqlite3.connect(path) as db: + db.executescript(""" + CREATE TABLE User(id INTEGER PRIMARY KEY,username TEXT); + CREATE TABLE sub_bind(key TEXT,user TEXT,user_id INTEGER); + CREATE TABLE sub_update_record(id INTEGER PRIMARY KEY,user_id INTEGER,time INTEGER); + INSERT INTO User VALUES(3,'alice'); + INSERT INTO sub_bind VALUES('public-test-key','alice',3); + """) + dry = migration.migrate(path) + self.assertEqual(2, len(dry["tables_to_create"])) + with self.assertRaises(ValueError): + migration.migrate(path, True) + first = migration.migrate(path, True, True) + self.assertEqual("ok", first["integrity"]) + self.assertEqual(0o600, Path(first["backup"]).stat().st_mode & 0o777) + with sqlite3.connect(first["backup"]) as db: + self.assertEqual(0, db.execute("SELECT count(*) FROM sqlite_master WHERE name='subscription_event_binding'").fetchone()[0]) + with sqlite3.connect(path) as db: + row = db.execute("SELECT key_hash,user_id,user_name FROM subscription_event_binding").fetchone() + self.assertEqual((hashlib.sha256(b"public-test-key").hexdigest(), 3, "alice"), row) + db.execute("INSERT INTO subscription_event_receipt VALUES('id','hash','v2',0,'primary',1,'fingerprint','recorded')") + second = migration.migrate(path, True, True) + self.assertEqual([], second["tables_to_create"]) + with sqlite3.connect(path) as db: + self.assertEqual(1, db.execute("SELECT count(*) FROM subscription_event_receipt").fetchone()[0]) + self.assertEqual(1, db.execute("SELECT count(*) FROM subscription_event_binding").fetchone()[0]) + + def test_legacy_labels_without_user_ids_remain_null(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "main.db" + with sqlite3.connect(path) as db: + db.executescript(""" + CREATE TABLE User(id INTEGER PRIMARY KEY,username TEXT); + CREATE TABLE sub_bind(key TEXT,user TEXT,user_id INTEGER); + CREATE TABLE sub_update_record(id INTEGER PRIMARY KEY,user_id INTEGER,time INTEGER); + INSERT INTO sub_bind VALUES('legacy-key','legacy-label',NULL); + """) + result = migration.migrate(path, True, True) + self.assertEqual(1, result["history_bindings"]) + with sqlite3.connect(path) as db: + row = db.execute("SELECT key_hash,user_id,user_name FROM subscription_event_binding").fetchone() + self.assertEqual((hashlib.sha256(b"legacy-key").hexdigest(), None, "legacy-label"), row) + self.assertIsNone(db.execute("SELECT user_id FROM sub_bind").fetchone()[0]) + self.assertEqual(0, db.execute("SELECT count(*) FROM User").fetchone()[0]) + migration.migrate(path, True, True) + with sqlite3.connect(path) as db: + self.assertEqual(1, db.execute("SELECT count(*) FROM subscription_event_binding").fetchone()[0]) + + def test_orphan_binding_aborts_without_ddl(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "main.db" + with sqlite3.connect(path) as db: + db.executescript(""" + CREATE TABLE User(id INTEGER PRIMARY KEY,username TEXT); + CREATE TABLE sub_bind(key TEXT,user TEXT,user_id INTEGER); + CREATE TABLE sub_update_record(id INTEGER PRIMARY KEY,user_id INTEGER,time INTEGER); + INSERT INTO sub_bind VALUES('test','deleted',9); + """) + with self.assertRaises(ValueError): + migration.migrate(path, True, True) + with sqlite3.connect(path) as db: + self.assertEqual(0, db.execute("SELECT count(*) FROM sqlite_master WHERE name LIKE 'subscription_event_%'").fetchone()[0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/main/java/com/lion/lionwebsite/Controller/SubscriptionEventController.java b/src/main/java/com/lion/lionwebsite/Controller/SubscriptionEventController.java new file mode 100644 index 0000000..b6c8f55 --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Controller/SubscriptionEventController.java @@ -0,0 +1,33 @@ +package com.lion.lionwebsite.Controller; + +import com.lion.lionwebsite.Service.SubscriptionEventService; +import jakarta.servlet.http.HttpServletRequest; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RestController; + +import java.io.IOException; +import java.util.Map; + +@RestController +public class SubscriptionEventController { + private final SubscriptionEventService events; + + public SubscriptionEventController(SubscriptionEventService events) { this.events = events; } + + @PostMapping(value = "/internal/subscription-events", produces = "application/json") + public ResponseEntity> ingest(HttpServletRequest request) throws IOException { + // Bound actual bytes even for chunked requests and dishonest Content-Length headers. + byte[] body = request.getInputStream().readNBytes(SubscriptionEventService.MAX_BODY + 1); + if (body.length > SubscriptionEventService.MAX_BODY) + return ResponseEntity.badRequest().body(Map.of("error", "invalid subscription events")); + if (!events.verify(body, request.getHeader("X-Subscription-Timestamp"), request.getHeader("X-Subscription-Signature"))) + return ResponseEntity.status(403).body(Map.of("error", "authentication failed")); + try { + return ResponseEntity.ok(events.ingest(events.parseBatch(body))); + } catch (IllegalArgumentException invalid) { + return ResponseEntity.badRequest().body(Map.of("error", "invalid subscription events")); + } + // All infrastructure failures propagate as 5xx; never acknowledge an uncommitted batch. + } +} diff --git a/src/main/java/com/lion/lionwebsite/Dao/normal/SubMapper.java b/src/main/java/com/lion/lionwebsite/Dao/normal/SubMapper.java index 6c19ba2..c7606cc 100644 --- a/src/main/java/com/lion/lionwebsite/Dao/normal/SubMapper.java +++ b/src/main/java/com/lion/lionwebsite/Dao/normal/SubMapper.java @@ -80,7 +80,7 @@ public interface SubMapper { @Select("select count(user) from sub_update_record where user_id=(select id from User where username=#{user}) or (user_id is null and user=#{user})") Integer selectUpdateRecordCount(String user); - @Select("select min(id) from sub_update_record where user_id=(select id from User where username=#{user}) or (user_id is null and user=#{user})") + @Select("select id from sub_update_record where user_id=(select id from User where username=#{user}) or (user_id is null and user=#{user}) order by time asc,id asc limit 1") Integer selectMinUpdateRecordId(String user); /** 后台补齐归属地：只更新该用户最近一条记录，避免覆盖更早的历史记录。 */ diff --git a/src/main/java/com/lion/lionwebsite/Dao/normal/SubscriptionEventMapper.java b/src/main/java/com/lion/lionwebsite/Dao/normal/SubscriptionEventMapper.java new file mode 100644 index 0000000..2420691 --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Dao/normal/SubscriptionEventMapper.java @@ -0,0 +1,31 @@ +package com.lion.lionwebsite.Dao.normal; + +import com.lion.lionwebsite.Domain.SubscriptionEvent; +import com.lion.lionwebsite.Domain.SubUpdateRecord; +import com.lion.lionwebsite.Domain.SubscriptionEventBinding; +import org.apache.ibatis.annotations.*; + +@Mapper +public interface SubscriptionEventMapper { + @Select("select fingerprint from subscription_event_receipt where event_id=#{eventId}") + String fingerprint(String eventId); + + @Insert("insert into subscription_event_receipt(event_id,key_hash,client,occurred_at,source,ingested_at,fingerprint,outcome) values(#{event.eventId},#{event.keyHash},#{event.client},#{event.occurredAt},#{source},#{ingestedAt},#{fingerprint},#{outcome})") + void insertReceipt(@Param("event") SubscriptionEvent event, @Param("source") String source, + @Param("ingestedAt") long ingestedAt, @Param("fingerprint") String fingerprint, + @Param("outcome") String outcome); + + @Select("select key_hash as keyHash,user_id as userId,user_name as userName from subscription_event_binding where key_hash=#{hash}") + SubscriptionEventBinding binding(String hash); + + @Insert("insert into subscription_event_binding(key_hash,user_id,user_name,first_seen_at,last_seen_at) values(#{hash},#{userId},#{userName},#{now},#{now}) on conflict(key_hash) do update set user_name=excluded.user_name,last_seen_at=excluded.last_seen_at where subscription_event_binding.user_id IS excluded.user_id") + void rememberBinding(@Param("hash") String hash, @Param("userId") Integer userId, + @Param("userName") String userName, @Param("now") long now); + + @Insert("insert into sub_update_record(user,ip,UA,time,location,user_id) values(#{user},#{ip},#{UA},#{time},#{location},#{userId})") + @Options(useGeneratedKeys = true, keyProperty = "id") + void insertRecord(SubUpdateRecord record); + + @Delete("delete from sub_update_record where (user_id=#{userId} or (user_id is null and user=#{userName})) and id not in (select id from sub_update_record where user_id=#{userId} or (user_id is null and user=#{userName}) order by time desc,id desc limit 10)") + void pruneRecords(@Param("userId") Integer userId, @Param("userName") String userName); +} diff --git a/src/main/java/com/lion/lionwebsite/Domain/SubscriptionEvent.java b/src/main/java/com/lion/lionwebsite/Domain/SubscriptionEvent.java new file mode 100644 index 0000000..95ec46a --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Domain/SubscriptionEvent.java @@ -0,0 +1,16 @@ +package com.lion.lionwebsite.Domain; + +import lombok.AllArgsConstructor; +import lombok.Data; + +/** Validated, authenticated event. Never contains the public subscription key. */ +@Data +@AllArgsConstructor +public class SubscriptionEvent { + private String eventId; + private String keyHash; + private String client; + private String clientIp; + private String userAgent; + private long occurredAt; +} diff --git a/src/main/java/com/lion/lionwebsite/Domain/SubscriptionEventBinding.java b/src/main/java/com/lion/lionwebsite/Domain/SubscriptionEventBinding.java new file mode 100644 index 0000000..3c72c6c --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Domain/SubscriptionEventBinding.java @@ -0,0 +1,10 @@ +package com.lion.lionwebsite.Domain; + +import lombok.Data; + +@Data +public class SubscriptionEventBinding { + private String keyHash; + private Integer userId; + private String userName; +} diff --git a/src/main/java/com/lion/lionwebsite/Service/SubService.java b/src/main/java/com/lion/lionwebsite/Service/SubService.java index ae01413..35742d1 100644 --- a/src/main/java/com/lion/lionwebsite/Service/SubService.java +++ b/src/main/java/com/lion/lionwebsite/Service/SubService.java @@ -5,6 +5,7 @@ import com.lion.lionwebsite.Dao.normal.SubMapper; import com.lion.lionwebsite.Dao.normal.UserMapper; import com.lion.lionwebsite.Domain.SubBind; import com.lion.lionwebsite.Domain.SubUpdateRecord; +import com.lion.lionwebsite.Domain.SubscriptionEvent; import com.lion.lionwebsite.Domain.SubscriptionAccount; import com.lion.lionwebsite.Util.CustomUtil; import com.lion.lionwebsite.Util.FileDownload; @@ -47,6 +48,9 @@ import java.util.function.Supplier; @Qualifier("mainTransactionTemplate") TransactionTemplate transactions; + @Autowired(required = false) + SubscriptionEventService subscriptionEvents; + /** * 归属地查询是同步外呼（最长十几秒），且失败只影响记录里的一列展示。 * 若放在订阅分发路径上，用户取一次订阅就得先等它完成，因此改为后台补齐。 @@ -188,6 +192,7 @@ import java.util.function.Supplier; String key = RandomUtil.randomString(8); while (subMapper.selectSubBindExist(key)) key = RandomUtil.randomString(8); + rememberEventBindings(); subMapper.updateSubBindKey(user, key); subMapper.deleteSubUpdateRecord(user); remoteService.requestSubscriptionSync(); @@ -239,8 +244,9 @@ import java.util.function.Supplier; } String ip = resolveClientIp(request); String ua = request.getHeader("User-Agent"); - if (ua == null) return; - recordUpdate(subBind, ip, ua); + SubscriptionEvent event = subscriptionEvents == null ? null : subscriptionEvents.requestEvent(request, key, client); + if (ua == null && event == null) return; + if (event == null) recordUpdate(subBind, ip, ua); if (!"v2".equals(client) && !"cat".equals(client)) { sendStatus(response, HttpServletResponse.SC_BAD_REQUEST, "client error"); return; @@ -250,6 +256,8 @@ import java.util.function.Supplier; sendStatus(response, HttpServletResponse.SC_SERVICE_UNAVAILABLE, "subscription unavailable"); return; } + if (event != null && !"HEAD".equalsIgnoreCase(request.getMethod())) + subscriptionEvents.recordPrimary(event, subBind); FileDownload.export(request, response, path.toString()); } @@ -329,12 +337,17 @@ import java.util.function.Supplier; } private String deleteSubBindUnlocked(String user) { + rememberEventBindings(); subMapper.deleteSubBind(user); subMapper.deleteSubUpdateRecord(user); remoteService.requestSubscriptionSync(); return Response._success("删除成功"); } + private void rememberEventBindings() { + if (subscriptionEvents != null) subscriptionEvents.rememberBindings(subMapper.selectAllSubBind()); + } + private String withWriteLock(Supplier action) { Lock stateLock = stateCoordinator.writeLock(); stateLock.lock(); diff --git a/src/main/java/com/lion/lionwebsite/Service/SubscriptionEventService.java b/src/main/java/com/lion/lionwebsite/Service/SubscriptionEventService.java new file mode 100644 index 0000000..7e8f320 --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Service/SubscriptionEventService.java @@ -0,0 +1,222 @@ +package com.lion.lionwebsite.Service; + +import com.lion.lionwebsite.Dao.normal.SubMapper; +import com.lion.lionwebsite.Dao.normal.SubscriptionEventMapper; +import com.lion.lionwebsite.Dao.normal.UserMapper; +import com.lion.lionwebsite.Domain.*; +import com.lion.lionwebsite.Util.CustomUtil; +import jakarta.annotation.PreDestroy; +import jakarta.servlet.http.HttpServletRequest; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; +import org.springframework.transaction.support.TransactionTemplate; +import tools.jackson.databind.JsonNode; + +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.time.Clock; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.locks.Lock; + +@Service +public class SubscriptionEventService { + public static final int MAX_BODY = 256 * 1024; + private final SubscriptionEventMapper events; + private final SubMapper subscriptions; + private final UserMapper users; + private final TransactionTemplate transactions; + private final SubscriptionStateCoordinator coordinator; + @Value("${subscription.standby.sync-secret:}") String syncSecret; + Clock clock = Clock.systemUTC(); + private final ExecutorService locations = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(128), runnable -> { + Thread thread = new Thread(runnable, "subscription-event-location"); + thread.setDaemon(true); + return thread; + }, new ThreadPoolExecutor.AbortPolicy()); + + public SubscriptionEventService(SubscriptionEventMapper events, SubMapper subscriptions, UserMapper users, + @Qualifier("mainTransactionTemplate") TransactionTemplate transactions, + SubscriptionStateCoordinator coordinator) { + this.events = events; + this.subscriptions = subscriptions; + this.users = users; + this.transactions = transactions; + this.coordinator = coordinator; + } + + /** Auth failure deliberately discloses neither signature nor payload. */ + public boolean verify(byte[] body, String timestamp, String signature) { + if (body == null || body.length > MAX_BODY || syncSecret == null || syncSecret.isBlank() + || timestamp == null || !timestamp.matches("[0-9]{1,12}") + || signature == null || !signature.matches("[0-9a-fA-F]{64}")) return false; + try { + long seconds = Long.parseLong(timestamp); + if (Math.abs(clock.instant().getEpochSecond() - seconds) > 300) return false; + Mac mac = Mac.getInstance("HmacSHA256"); + mac.init(new SecretKeySpec(syncSecret.getBytes(StandardCharsets.UTF_8), "HmacSHA256")); + byte[] expected = mac.doFinal(("v1\n" + timestamp + "\n" + sha256(body)).getBytes(StandardCharsets.UTF_8)); + return MessageDigest.isEqual(expected, HexFormat.of().parseHex(signature)); + } catch (Exception invalid) { return false; } + } + + /** Invalid or absent signed headers fall back to the ordinary logging path. */ + public SubscriptionEvent requestEvent(HttpServletRequest request, String key, String client) { + String encoded = request.getHeader("X-Subscription-Event"); + if (encoded == null || encoded.length() > 8192 || !encoded.matches("[A-Za-z0-9_-]+")) return null; + try { + byte[] body = Base64.getUrlDecoder().decode(encoded); + if (!verify(body, request.getHeader("X-Subscription-Timestamp"), request.getHeader("X-Subscription-Signature"))) return null; + SubscriptionEvent event = parse(CustomUtil.objectMapper.readTree(body)); + String ua = Objects.requireNonNullElse(request.getHeader("User-Agent"), ""); + if (!event.getKeyHash().equals(sha256(key.getBytes(StandardCharsets.UTF_8))) + || !event.getClient().equals(client) || !event.getUserAgent().equals(ua)) return null; + return event; + } catch (Exception invalid) { return null; } + } + + public List parseBatch(byte[] body) { + try { + if (body == null || body.length > MAX_BODY) throw new IllegalArgumentException(); + JsonNode root = CustomUtil.objectMapper.readTree(body); + JsonNode list = root.get("events"); + if (!root.isObject() || list == null || !list.isArray() || list.size() > 100) throw new IllegalArgumentException(); + List result = new ArrayList<>(); + for (JsonNode item : list) result.add(parse(item)); + return result; + } catch (Exception invalid) { throw new IllegalArgumentException("invalid subscription events"); } + } + + SubscriptionEvent parse(JsonNode node) { + if (node == null || !node.isObject()) throw new IllegalArgumentException(); + String id = text(node, "eventId", 36); + if (!UUID.fromString(id).toString().equals(id)) throw new IllegalArgumentException(); + String hash = text(node, "keyHash", 64); + if (!hash.matches("[0-9a-f]{64}")) throw new IllegalArgumentException(); + String client = text(node, "client", 3); + if (!client.equals("v2") && !client.equals("cat")) throw new IllegalArgumentException(); + String ip = text(node, "clientIp", 64); + // Literal IPv4/IPv6 only: no host names, URL fragments or query injection into geolocation. + if (ip.isEmpty() || !ip.matches("[0-9A-Fa-f:.]+")) throw new IllegalArgumentException(); + String ua = text(node, "userAgent", 2048); + if (ua.chars().anyMatch(c -> c < 32 || c == 127)) throw new IllegalArgumentException(); + JsonNode time = node.get("occurredAt"); + if (time == null || !time.isIntegralNumber() || !time.canConvertToLong()) throw new IllegalArgumentException(); + long occurred = time.longValue(); + if (occurred < 0 || occurred > clock.millis() + 300_000) throw new IllegalArgumentException(); + return new SubscriptionEvent(id, hash, client, ip, ua, occurred); + } + + private static String text(JsonNode node, String field, int max) { + JsonNode value = node.get(field); + if (value == null || !value.isString() || value.stringValue().length() > max) throw new IllegalArgumentException(); + return value.stringValue(); + } + + public Map ingest(List batch) { + Lock lock = coordinator.readLock(); + lock.lock(); + try { + // Return acknowledgements only after execute has returned and the entire transaction committed. + List acknowledged = transactions.execute(status -> { + List ids = new ArrayList<>(); + for (SubscriptionEvent event : batch) if (persist(event, "standby", null)) ids.add(event.getEventId()); + return ids; + }); + return Map.of("acknowledged", Objects.requireNonNull(acknowledged)); + } finally { lock.unlock(); } + } + + public void recordPrimary(SubscriptionEvent event, SubBind binding) { + transactions.executeWithoutResult(status -> { + if (!persist(event, "primary", binding)) throw new IllegalStateException("subscription user unavailable"); + }); + } + + private boolean persist(SubscriptionEvent event, String source, SubBind current) { + String fingerprint = sha256((event.getEventId() + "\n" + event.getKeyHash() + "\n" + event.getClient() + + "\n" + event.getOccurredAt() + "\n" + event.getClientIp() + "\n" + event.getUserAgent()).getBytes(StandardCharsets.UTF_8)); + String old = events.fingerprint(event.getEventId()); + if (old != null) { + if (!old.equals(fingerprint)) throw new IllegalArgumentException("event ID conflict"); + return true; + } + Integer userId = current == null ? null : current.getUserId(); + String userName = current == null ? null : current.getUser(); + if (current == null) { + boolean found = false; + for (SubBind binding : subscriptions.selectAllSubBind()) { + if (sha256(binding.getKey().getBytes(StandardCharsets.UTF_8)).equals(event.getKeyHash())) { + userId = binding.getUserId(); + userName = binding.getUser(); + found = true; + break; + } + } + if (!found) { + SubscriptionEventBinding binding = events.binding(event.getKeyHash()); + if (binding == null) return false; + userId = binding.getUserId(); + userName = binding.getUserName(); + } + } + if (userId != null) { + User user = users.selectUserById(userId); + if (user == null) { + events.insertReceipt(event, source, clock.millis(), fingerprint, "user_deleted"); + return true; + } + userName = user.getUsername(); + } + // NULL user_id is a valid legacy subscription label, not a missing/deleted User. + // Keep it NULL even if someone later creates a User with the same label. + if (userName == null || userName.isBlank()) return false; + events.rememberBinding(event.getKeyHash(), userId, userName, clock.millis()); + SubUpdateRecord record = new SubUpdateRecord(0, userName, event.getClientIp(), event.getUserAgent(), + new Date(event.getOccurredAt()), "unknown", userId); + events.insertRecord(record); + events.pruneRecords(userId, userName); + events.insertReceipt(event, source, clock.millis(), fingerprint, "recorded"); + // Never perform network calls before commit. A dropped/rolled-back record has no enrichment task. + org.springframework.transaction.support.TransactionSynchronizationManager.registerSynchronization( + new org.springframework.transaction.support.TransactionSynchronization() { + @Override public void afterCommit() { enrich(record.getId(), event.getClientIp()); } + }); + return true; + } + + public void rememberBindings(Collection bindings) { + for (SubBind binding : bindings) { + Integer userId = binding.getUserId(); + String name = binding.getUser(); + if (userId != null) { + User user = users.selectUserById(userId); + if (user != null) name = user.getUsername(); + } + if (name != null && !name.isBlank()) events.rememberBinding( + sha256(binding.getKey().getBytes(StandardCharsets.UTF_8)), userId, name, clock.millis()); + } + } + + private void enrich(int recordId, String ip) { + if (recordId <= 0) return; + try { locations.execute(() -> { + String location = SubService.resolveLocation(ip); + if (location != null && !location.isBlank() && !location.equals("unknown")) { + try { subscriptions.updateSubUpdateRecordLocation(recordId, location); } + catch (Exception ignored) { /* Optional display enrichment; no request failure or sensitive log. */ } + } + }); } catch (RejectedExecutionException ignored) { } + } + + public static String sha256(byte[] body) { + try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(body)); } + catch (Exception failure) { throw new IllegalStateException("SHA-256 unavailable"); } + } + + @PreDestroy void close() { locations.shutdownNow(); } +} diff --git a/src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java b/src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java index 0843b0c..e070e53 100644 --- a/src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java +++ b/src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java @@ -9,6 +9,7 @@ import com.lion.lionwebsite.Util.CustomUtil; import com.lion.lionwebsite.Util.SubscriptionCacheStore; import lombok.RequiredArgsConstructor; import org.springframework.beans.factory.annotation.Value; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import jakarta.annotation.PostConstruct; @@ -29,6 +30,9 @@ public class SubscriptionStandbySnapshotService { private final SubscriptionStateCoordinator stateCoordinator; private final ObjectMapper objectMapper = CustomUtil.objectMapper; + @Autowired(required = false) + SubscriptionEventService subscriptionEvents; + @Value("${subscription.standby.sync-secret:}") String syncSecret; @@ -78,7 +82,9 @@ public class SubscriptionStandbySnapshotService { List accounts = new ArrayList<>(accountMap.values()); accounts.sort(Comparator.comparing(SubscriptionAccountSnapshot::getAccountId)); List bindings = new ArrayList<>(); - for (SubBind bind : subMapper.selectAllSubBind()) { + List currentBindings = subMapper.selectAllSubBind(); + if (subscriptionEvents != null) subscriptionEvents.rememberBindings(currentBindings); + for (SubBind bind : currentBindings) { if (bind.getSubscriptionAccountId() == null || !accountMap.containsKey(bind.getSubscriptionAccountId())) continue; SubscriptionBindingSnapshot snapshot = new SubscriptionBindingSnapshot(); diff --git a/src/test/java/com/lion/lionwebsite/Configuration/MyBatisNativeConfigurationTest.java b/src/test/java/com/lion/lionwebsite/Configuration/MyBatisNativeConfigurationTest.java index 92145de..83b8b15 100644 --- a/src/test/java/com/lion/lionwebsite/Configuration/MyBatisNativeConfigurationTest.java +++ b/src/test/java/com/lion/lionwebsite/Configuration/MyBatisNativeConfigurationTest.java @@ -231,6 +231,22 @@ class MyBatisNativeConfigurationTest { "同名 XML 映射文件应登记"); } + @Test + void subscriptionEventMapperRegistersEventAndBindingReflection() { + var processor = new MyBatisNativeConfiguration.MyBatisBeanFactoryInitializationAotProcessor(); + var factory = new org.springframework.beans.factory.support.DefaultListableBeanFactory(); + var definition = new RootBeanDefinition(MapperFactoryBean.class); + definition.getPropertyValues().add("mapperInterface", com.lion.lionwebsite.Dao.normal.SubscriptionEventMapper.class); + factory.registerBeanDefinition("subscriptionEventMapper", definition); + var hints = new RuntimeHints(); + var contribution = processor.processAheadOfTime(factory); + assertNotNull(contribution); + contribution.applyTo(new StubGenerationContext(hints), new NoOpInitializationCode()); + assertTrue(RuntimeHintsPredicates.reflection().onType(com.lion.lionwebsite.Domain.SubscriptionEvent.class).test(hints)); + assertTrue(RuntimeHintsPredicates.reflection().onType(com.lion.lionwebsite.Domain.SubscriptionEventBinding.class).test(hints)); + assertTrue(RuntimeHintsPredicates.proxies().forInterfaces(com.lion.lionwebsite.Dao.normal.SubscriptionEventMapper.class).test(hints)); + } + /** applyTo 只用 runtimeHints，初始化代码侧给一个记录型空实现即可。 */ private static final class NoOpInitializationCode implements org.springframework.beans.factory.aot.BeanFactoryInitializationCode { diff --git a/src/test/java/com/lion/lionwebsite/Service/SubscriptionEventIntegrationTest.java b/src/test/java/com/lion/lionwebsite/Service/SubscriptionEventIntegrationTest.java new file mode 100644 index 0000000..c6e8ba8 --- /dev/null +++ b/src/test/java/com/lion/lionwebsite/Service/SubscriptionEventIntegrationTest.java @@ -0,0 +1,317 @@ +package com.lion.lionwebsite.Service; + +import com.lion.lionwebsite.Controller.SubscriptionEventController; +import com.lion.lionwebsite.Dao.normal.SubMapper; +import com.lion.lionwebsite.Dao.normal.SubscriptionEventMapper; +import com.lion.lionwebsite.Dao.normal.UserMapper; +import com.lion.lionwebsite.Domain.SubscriptionEvent; +import com.lion.lionwebsite.Util.CustomUtil; +import org.junit.jupiter.api.*; +import org.junit.jupiter.api.io.TempDir; +import org.mybatis.spring.SqlSessionFactoryBean; +import org.mybatis.spring.SqlSessionTemplate; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.DriverManagerDataSource; +import org.springframework.mock.web.MockHttpServletRequest; +import org.springframework.mock.web.MockHttpServletResponse; +import java.nio.file.Files; +import static org.mockito.Mockito.*; +import org.springframework.transaction.support.TransactionTemplate; + +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.*; + +import static org.junit.jupiter.api.Assertions.*; + +/** Real SQLite/MyBatis/Spring transactions: these assertions cover committed database state. */ +class SubscriptionEventIntegrationTest { + @TempDir Path directory; + private JdbcTemplate db; + private SubscriptionEventService service; + private SubMapper subscriptions; + private static final long NOW = 1_792_555_200_000L; + private static final String SECRET = "integration-test-only-secret"; + private static final String KEY = "example-key"; + + @BeforeEach void setUp() throws Exception { + var source = new DriverManagerDataSource("jdbc:sqlite:" + directory.resolve("events.db") + "?busy_timeout=10000"); + db = new JdbcTemplate(source); + db.execute("create table User(id integer primary key,username text,AuthCode text,isEnable integer,lastAccessTime text)"); + db.execute("create table subscription_account(id integer primary key,name text,enabled integer,filter_high_multiplier integer)"); + db.execute("create table sub_bind(key text primary key,user text,user_id integer,subscription_account_id integer)"); + db.execute("create table sub_update_record(id integer primary key autoincrement,user text,ip text,UA text,time integer,location text,user_id integer)"); + db.execute("create table subscription_event_receipt(event_id text primary key,key_hash text,client text,occurred_at integer,source text,ingested_at integer,fingerprint text,outcome text)"); + db.execute("create table subscription_event_binding(key_hash text primary key,user_id integer,user_name text,first_seen_at integer,last_seen_at integer)"); + db.update("insert into User(id,username,isEnable) values(5,'alice',1)"); + db.update("insert into subscription_account values(1,'account',1,1)"); + db.update("insert into sub_bind values(?,'alice',5,1)", KEY); + var bean = new SqlSessionFactoryBean(); + bean.setDataSource(source); + var factory = bean.getObject(); + factory.getConfiguration().addMapper(SubMapper.class); + factory.getConfiguration().addMapper(UserMapper.class); + factory.getConfiguration().addMapper(SubscriptionEventMapper.class); + var session = new SqlSessionTemplate(factory); + subscriptions = session.getMapper(SubMapper.class); + service = new SubscriptionEventService(session.getMapper(SubscriptionEventMapper.class), subscriptions, + session.getMapper(UserMapper.class), new TransactionTemplate(new DataSourceTransactionManager(source)), + new SubscriptionStateCoordinator()); + service.syncSecret = SECRET; + service.clock = Clock.fixed(Instant.ofEpochMilli(NOW), ZoneOffset.UTC); + service.close(); // Reject optional geolocation work; never make network calls from these tests. + } + + @Test void hmacRejectsInvalidExpiredModifiedAndOversizedInputs() throws Exception { + byte[] body = batch(event(NOW - 60_000)); + String timestamp = Long.toString(NOW / 1000); + assertTrue(service.verify(body, timestamp, sign(body, timestamp))); + assertFalse(service.verify(body, timestamp, "0".repeat(64))); + assertFalse(service.verify("changed".getBytes(StandardCharsets.UTF_8), timestamp, sign(body, timestamp))); + String expired = Long.toString(NOW / 1000 - 301); + assertFalse(service.verify(body, expired, sign(body, expired))); + assertFalse(service.verify(body, "9".repeat(1024), "0".repeat(64))); + assertFalse(service.verify(new byte[SubscriptionEventService.MAX_BODY + 1], timestamp, "0".repeat(64))); + } + + @Test void primaryAndStandbySameEventProduceOneRecordAndPreserveOriginalMetadata() { + SubscriptionEvent event = event(NOW - 86_400_000); + service.recordPrimary(event, subscriptions.selectSubBind(KEY)); + assertEquals(List.of(event.getEventId()), service.ingest(List.of(event)).get("acknowledged")); + assertEquals(1, count("sub_update_record")); + assertEquals(1, count("subscription_event_receipt")); + assertEquals("primary", db.queryForObject("select source from subscription_event_receipt", String.class)); + Map row = db.queryForMap("select * from sub_update_record"); + assertEquals(event.getOccurredAt(), ((Number) row.get("time")).longValue()); + assertEquals("2001:db8::7", row.get("ip")); + assertEquals("client/1.2", row.get("UA")); + assertEquals(5, ((Number) row.get("user_id")).intValue()); + } + + @Test void keyRotationAndRenameRetainStableIdentity() { + service.rememberBindings(subscriptions.selectAllSubBind()); + db.update("update sub_bind set key='rotated-key'"); + db.update("update User set username='renamed' where id=5"); + SubscriptionEvent old = event(NOW - 30_000); + assertEquals(List.of(old.getEventId()), service.ingest(List.of(old)).get("acknowledged")); + assertEquals("renamed", db.queryForObject("select user from sub_update_record", String.class)); + assertEquals(5, db.queryForObject("select user_id from sub_update_record", Integer.class)); + assertEquals(SubscriptionEventService.sha256(KEY.getBytes(StandardCharsets.UTF_8)), + db.queryForObject("select key_hash from subscription_event_binding", String.class)); + } + + @Test void legacyLabelWithoutUserHasPrimaryStandbyReplayAndRotatedKeyHistory() { + db.update("update sub_bind set user='legacy-label',user_id=NULL"); + db.update("delete from User"); + SubscriptionEvent primary = event(NOW - 10_000); + service.recordPrimary(primary, subscriptions.selectSubBind(KEY)); + assertEquals(List.of(primary.getEventId()), service.ingest(List.of(primary)).get("acknowledged")); + assertEquals(1, count("sub_update_record")); + assertEquals("legacy-label", db.queryForObject("select user from sub_update_record", String.class)); + assertNull(db.queryForObject("select user_id from sub_update_record", Integer.class)); + service.rememberBindings(subscriptions.selectAllSubBind()); + db.update("update sub_bind set key='rotated-legacy-key'"); + SubscriptionEvent offline = event(NOW - 5_000); + assertEquals(List.of(offline.getEventId()), service.ingest(List.of(offline)).get("acknowledged")); + assertEquals(2, count("sub_update_record")); + assertEquals(0, count("User")); + assertEquals(2, db.queryForObject("select count(*) from sub_update_record where user='legacy-label' and user_id is null", Integer.class)); + for (int i = 0; i < 12; i++) service.ingest(List.of(event(NOW - i))); + assertEquals(10, count("sub_update_record")); + assertEquals(14, count("subscription_event_receipt")); + service.ingest(List.of(primary)); + assertEquals(10, count("sub_update_record")); + } + + @Test void legacyLabelDoesNotAdoptNewSameNamedUser() { + db.update("update sub_bind set user='legacy-label',user_id=NULL"); + service.rememberBindings(subscriptions.selectAllSubBind()); + db.update("delete from sub_bind"); + db.update("insert into User(id,username,isEnable) values(8,'legacy-label',1)"); + SubscriptionEvent event = event(NOW); + assertEquals(List.of(event.getEventId()), service.ingest(List.of(event)).get("acknowledged")); + assertNull(db.queryForObject("select user_id from sub_update_record", Integer.class)); + } + + @Test void unknownBindingIsNotAcknowledgedAndDeletedUserIsTerminal() { + SubscriptionEvent unknown = event(NOW); + unknown.setKeyHash("0".repeat(64)); + assertEquals(List.of(), service.ingest(List.of(unknown)).get("acknowledged")); + assertEquals(0, count("subscription_event_receipt")); + service.rememberBindings(subscriptions.selectAllSubBind()); + db.update("delete from sub_bind"); + db.update("delete from User"); + SubscriptionEvent deleted = event(NOW - 1_000); + assertEquals(List.of(deleted.getEventId()), service.ingest(List.of(deleted)).get("acknowledged")); + assertEquals("user_deleted", db.queryForObject("select outcome from subscription_event_receipt", String.class)); + assertEquals(0, count("sub_update_record")); + } + + @Test void transactionFailureRollsBackWholeBatchAndControllerNeverAcknowledges() throws Exception { + db.execute("create trigger fail_receipt before insert on subscription_event_receipt begin select raise(ABORT,'test rollback'); end"); + SubscriptionEvent event = event(NOW); + assertThrows(RuntimeException.class, () -> service.ingest(List.of(event))); + assertEquals(0, count("sub_update_record")); + assertEquals(0, count("subscription_event_receipt")); + var controller = new SubscriptionEventController(service); + assertThrows(RuntimeException.class, () -> controller.ingest(signedPost(batch(event)))); + assertEquals(0, count("sub_update_record")); + } + + @Test void pruningUsesEventTimeAndReceiptsSurviveVisibleRecordDeletion() { + SubscriptionEvent oldest = event(NOW - 100_000); + service.ingest(List.of(oldest)); + List newer = new ArrayList<>(); + for (int i = 0; i < 12; i++) newer.add(event(NOW - i * 1_000)); + service.ingest(newer); + assertEquals(10, count("sub_update_record")); + assertEquals(13, count("subscription_event_receipt")); + assertEquals(NOW - 9_000, db.queryForObject("select min(time) from sub_update_record", Long.class)); + assertEquals(List.of(oldest.getEventId()), service.ingest(List.of(oldest)).get("acknowledged")); + assertEquals(10, count("sub_update_record")); + db.update("delete from sub_update_record"); + service.recordPrimary(newer.getFirst(), subscriptions.selectSubBind(KEY)); + assertEquals(0, count("sub_update_record")); + assertEquals(13, count("subscription_event_receipt")); + } + + @Test void signedGetVerifiesKeyClientUaAndUsesAuthenticatedIp() throws Exception { + SubscriptionEvent event = event(NOW); + byte[] body = CustomUtil.objectMapper.writeValueAsBytes(asMap(event)); + var request = new MockHttpServletRequest("GET", "/sub/cat/" + KEY); + request.setRemoteAddr("127.0.0.1"); + request.addHeader("X-Forwarded-For", "198.51.100.123"); + request.addHeader("User-Agent", event.getUserAgent()); + request.addHeader("X-Subscription-Event", Base64.getUrlEncoder().withoutPadding().encodeToString(body)); + String timestamp = Long.toString(NOW / 1000); + request.addHeader("X-Subscription-Timestamp", timestamp); + request.addHeader("X-Subscription-Signature", sign(body, timestamp)); + SubscriptionEvent parsed = service.requestEvent(request, KEY, "cat"); + assertNotNull(parsed); + assertEquals("2001:db8::7", parsed.getClientIp()); + assertNull(service.requestEvent(request, "wrong-key", "cat")); + assertNull(service.requestEvent(request, KEY, "v2")); + request.removeHeader("User-Agent"); + request.addHeader("User-Agent", "different-client"); + assertNull(service.requestEvent(request, KEY, "cat")); + } + + @Test void signedHeadAndUnavailableCacheDoNotWriteAndGetReplayWritesOnce() throws Exception { + var refresh = mock(SubscriptionRefreshService.class); + var sub = new SubService(subscriptions, mock(UserMapper.class), refresh, mock(RemoteService.class), + new SubscriptionStateCoordinator()); + sub.subscriptionEvents = service; + sub.locationExecutor = runnable -> { }; + Path cache = directory.resolve("subscription.txt"); + when(refresh.cachedPath(1, "cat")).thenReturn(cache); + SubscriptionEvent event = event(NOW); + var unavailable = new MockHttpServletResponse(); + sub.updateSub(unavailable, signedGet(event, "GET"), "cat", KEY); + assertEquals(503, unavailable.getStatus()); + assertEquals(0, count("sub_update_record")); + Files.writeString(cache, "subscription-content"); + var head = new MockHttpServletResponse(); + sub.updateSub(head, signedGet(event, "HEAD"), "cat", KEY); + assertEquals(200, head.getStatus()); + assertEquals(0, head.getContentAsByteArray().length); + assertEquals(0, count("sub_update_record")); + assertEquals(0, count("subscription_event_receipt")); + for (int i = 0; i < 2; i++) { + var get = new MockHttpServletResponse(); + sub.updateSub(get, signedGet(event, "GET"), "cat", KEY); + assertEquals(200, get.getStatus()); + assertEquals("subscription-content", get.getContentAsString()); + } + assertEquals(1, count("sub_update_record")); + assertEquals(1, count("subscription_event_receipt")); + assertEquals("2001:db8::7", db.queryForObject("select ip from sub_update_record", String.class)); + sub.shutdownLocationResolver(); + } + + @Test void invalidSignedGetStillUsesOrdinaryLogging() throws Exception { + var refresh = mock(SubscriptionRefreshService.class); + var sub = new SubService(subscriptions, mock(UserMapper.class), refresh, mock(RemoteService.class), + new SubscriptionStateCoordinator()); + sub.subscriptionEvents = service; + sub.locationExecutor = runnable -> { }; + Path cache = directory.resolve("subscription.txt"); + Files.writeString(cache, "subscription-content"); + when(refresh.cachedPath(1, "cat")).thenReturn(cache); + var request = signedGet(event(NOW), "GET"); + request.removeHeader("X-Subscription-Signature"); + request.addHeader("X-Subscription-Signature", "0".repeat(64)); + request.setRemoteAddr("198.51.100.10"); + sub.updateSub(new MockHttpServletResponse(), request, "cat", KEY); + assertEquals(1, count("sub_update_record")); + assertEquals(0, count("subscription_event_receipt")); + assertEquals("198.51.100.10", db.queryForObject("select ip from sub_update_record", String.class)); + sub.shutdownLocationResolver(); + } + + @Test void invalidInputAndAuthenticationReturnNoAcknowledgements() throws Exception { + var controller = new SubscriptionEventController(service); + var unauthenticated = new MockHttpServletRequest("POST", "/internal/subscription-events"); + unauthenticated.setContent(batch(event(NOW))); + assertEquals(403, controller.ingest(unauthenticated).getStatusCode().value()); + assertEquals(400, controller.ingest(signedPost("{}".getBytes(StandardCharsets.UTF_8))).getStatusCode().value()); + assertEquals(400, controller.ingest(signedPost(new byte[SubscriptionEventService.MAX_BODY + 1])).getStatusCode().value()); + SubscriptionEvent future = event(NOW + 300_001); + assertThrows(IllegalArgumentException.class, () -> service.parseBatch(batch(future))); + SubscriptionEvent tooLong = event(NOW); + tooLong.setUserAgent("a".repeat(2049)); + assertThrows(IllegalArgumentException.class, () -> service.parseBatch(batch(tooLong))); + assertEquals(0, count("subscription_event_receipt")); + } + + @Test void conflictingIdCannotAcknowledgeOrWriteChangedEvent() { + SubscriptionEvent original = event(NOW); + service.ingest(List.of(original)); + original.setClientIp("192.0.2.7"); + assertThrows(IllegalArgumentException.class, () -> service.ingest(List.of(original))); + assertEquals(1, count("sub_update_record")); + } + + private int count(String table) { return db.queryForObject("select count(*) from " + table, Integer.class); } + private SubscriptionEvent event(long time) { + return new SubscriptionEvent(UUID.randomUUID().toString(), SubscriptionEventService.sha256(KEY.getBytes(StandardCharsets.UTF_8)), + "cat", "2001:db8::7", "client/1.2", time); + } + private static Map asMap(SubscriptionEvent event) { + return Map.of("eventId", event.getEventId(), "keyHash", event.getKeyHash(), "client", event.getClient(), + "clientIp", event.getClientIp(), "userAgent", event.getUserAgent(), "occurredAt", event.getOccurredAt()); + } + private static byte[] batch(SubscriptionEvent event) { + return CustomUtil.objectMapper.writeValueAsBytes(Map.of("events", List.of(asMap(event)))); + } + private static String sign(byte[] body, String timestamp) throws Exception { + var mac = Mac.getInstance("HmacSHA256"); + mac.init(new SecretKeySpec(SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256")); + return HexFormat.of().formatHex(mac.doFinal(("v1\n" + timestamp + "\n" + SubscriptionEventService.sha256(body)).getBytes(StandardCharsets.UTF_8))); + } + private MockHttpServletRequest signedGet(SubscriptionEvent event, String method) throws Exception { + byte[] body = CustomUtil.objectMapper.writeValueAsBytes(asMap(event)); + var request = new MockHttpServletRequest(method, "/sub/cat/" + KEY); + request.setRemoteAddr("127.0.0.1"); + request.addHeader("User-Agent", event.getUserAgent()); + request.addHeader("X-Subscription-Event", Base64.getUrlEncoder().withoutPadding().encodeToString(body)); + String timestamp = Long.toString(NOW / 1000); + request.addHeader("X-Subscription-Timestamp", timestamp); + request.addHeader("X-Subscription-Signature", sign(body, timestamp)); + return request; + } + + private MockHttpServletRequest signedPost(byte[] body) throws Exception { + var request = new MockHttpServletRequest("POST", "/internal/subscription-events"); + request.setContent(body); + String timestamp = Long.toString(NOW / 1000); + request.addHeader("X-Subscription-Timestamp", timestamp); + request.addHeader("X-Subscription-Signature", sign(body, timestamp)); + return request; + } +}