Add complimentary OOBE polish and configurable admin mTLS
Allow one server-audited onboarding polish request without credits and make the certificate gate temporarily reversible while preserving application authentication.
This commit is contained in:
@@ -31,6 +31,8 @@ IDENTITY_TOMBSTONE_RETENTION_DAYS=365
|
||||
# Admin console. Enable bootstrap for the first successful startup only, then
|
||||
# set it back to false and remove all four ADMIN_BOOTSTRAP_* credential values.
|
||||
ADMIN_ENABLED=false
|
||||
# Keep true by default. Set false only for a deliberate temporary mTLS bypass.
|
||||
ADMIN_MTLS_REQUIRED=true
|
||||
ADMIN_BOOTSTRAP_ENABLED=false
|
||||
ADMIN_BOOTSTRAP_OPERATOR_ID=replace-with-random-uuid
|
||||
ADMIN_BOOTSTRAP_USERNAME=owner
|
||||
|
||||
@@ -32,6 +32,7 @@ services:
|
||||
IDENTITY_TOMBSTONE_RETENTION_DAYS: ${IDENTITY_TOMBSTONE_RETENTION_DAYS:-365}
|
||||
|
||||
ADMIN_ENABLED: ${ADMIN_ENABLED:-false}
|
||||
ADMIN_MTLS_REQUIRED: ${ADMIN_MTLS_REQUIRED:-true}
|
||||
ADMIN_BOOTSTRAP_ENABLED: ${ADMIN_BOOTSTRAP_ENABLED:-false}
|
||||
ADMIN_BOOTSTRAP_OPERATOR_ID: ${ADMIN_BOOTSTRAP_OPERATOR_ID:-}
|
||||
ADMIN_BOOTSTRAP_USERNAME: ${ADMIN_BOOTSTRAP_USERNAME:-}
|
||||
|
||||
+24
-6
@@ -2,8 +2,21 @@
|
||||
|
||||
`account.osglab.com` 在同一个 TLS `server` 中同时承载移动端 API 和管理端。
|
||||
由于 TLS 握手发生在 HTTP 路径匹配之前,配置必须使用 server 级
|
||||
`ssl_verify_client optional`:普通客户端不提供证书时仍可正常访问,只有
|
||||
`/admin`、`/admin/`、`/v1/admin` 和其子路径要求验证成功。
|
||||
`ssl_verify_client optional`:普通客户端不提供证书时仍可正常访问。OpenResty 会把实际
|
||||
证书验证结果传给 Ktor;`ADMIN_MTLS_REQUIRED=true`(默认值)时,`/admin`、`/admin/`、
|
||||
`/v1/admin` 和其子路径要求验证成功。
|
||||
|
||||
## 临时关闭
|
||||
|
||||
在 1Panel/Compose 环境中显式设置并重启应用:
|
||||
|
||||
```text
|
||||
ADMIN_MTLS_REQUIRED=false
|
||||
```
|
||||
|
||||
关闭后,管理端无需客户端证书,但登录仍要求用户名、密码和 TOTP,其他会话、CSRF、RBAC、
|
||||
失败锁定、限流与审计规则保持不变。恢复时将该值改回 `true` 并重启应用。不要为了临时关闭
|
||||
而删除客户端 CA、证书或轮换记录。
|
||||
|
||||
## CA 与证书
|
||||
|
||||
@@ -22,14 +35,15 @@
|
||||
|
||||
## 上游信任边界
|
||||
|
||||
OpenResty 仅在管理路径且 `$ssl_client_verify = SUCCESS` 时向 Ktor 设置固定头:
|
||||
OpenResty 在管理路径用 `$ssl_client_verify` 覆盖并转发证书验证结果。有效证书对应:
|
||||
|
||||
```text
|
||||
X-OSG-mTLS-Verified: SUCCESS
|
||||
```
|
||||
|
||||
客户端传入的同名头会被覆盖;其他路径会删除该头。Ktor 只能把这个头作为“边缘已验证”
|
||||
信号,不能信任客户端提供的证书相关头,也不能用 DN、CN 或证书正文做隐式授权。
|
||||
无证书时该值为 `NONE`。客户端传入的同名头会被覆盖;其他路径会删除该头。Ktor 只能把
|
||||
OpenResty 写入的这个头作为“边缘已验证”信号,不能信任客户端提供的证书相关头,也不能用
|
||||
DN、CN 或证书正文做隐式授权。
|
||||
后端端口必须继续只监听 `127.0.0.1:18080`,否则攻击者可绕过边缘伪造该头。
|
||||
mTLS 只证明客户端持有受信证书,管理接口仍应执行应用层身份认证、授权和审计。
|
||||
|
||||
@@ -52,7 +66,7 @@ ADMIN_BOOTSTRAP_ENABLED=true
|
||||
|
||||
## 验证
|
||||
|
||||
将测试域名解析到目标边缘后执行:
|
||||
`ADMIN_MTLS_REQUIRED=true` 时,将测试域名解析到目标边缘后执行:
|
||||
|
||||
```sh
|
||||
# 无证书:管理路径必须是 404。
|
||||
@@ -74,6 +88,10 @@ curl -i -H 'X-OSG-mTLS-Verified: SUCCESS' \
|
||||
还应使用由非管理 CA 签发或已过期的客户端证书确认返回 404,并在 Ktor 测试端点确认:
|
||||
管理请求只收到固定值 `SUCCESS`,普通 API 不收到 `X-OSG-mTLS-Verified`。
|
||||
|
||||
`ADMIN_MTLS_REQUIRED=false` 时,无证书访问 `/admin/` 应返回管理页面,
|
||||
`/v1/admin/auth/session` 应返回匿名会话状态。伪造 `X-OSG-mTLS-Verified: SUCCESS` 不会
|
||||
改变结果,因为 OpenResty 会将其覆盖为实际验证状态。
|
||||
|
||||
部署后可在受信设备运行不含登录凭据的自动验收:
|
||||
|
||||
```sh
|
||||
|
||||
@@ -52,13 +52,10 @@ server {
|
||||
return 404;
|
||||
}
|
||||
|
||||
# Administrative endpoints are indistinguishable from missing routes unless
|
||||
# OpenResty verified a certificate issued by the dedicated admin client CA.
|
||||
# Always forward administrative paths to Ktor. Ktor decides whether mTLS is
|
||||
# required from ADMIN_MTLS_REQUIRED, while this edge overwrites the trust
|
||||
# signal so clients cannot spoof successful certificate verification.
|
||||
location ~ ^/(?:admin|v1/admin)(?:/|$) {
|
||||
if ($ssl_client_verify != SUCCESS) {
|
||||
return 404;
|
||||
}
|
||||
|
||||
client_max_body_size 32k;
|
||||
limit_req zone=account_api burst=40 nodelay;
|
||||
# Defining a location-level header disables inheritance from the server
|
||||
@@ -76,8 +73,7 @@ server {
|
||||
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
||||
proxy_set_header X-Forwarded-Proto https;
|
||||
proxy_set_header X-Request-ID $request_id;
|
||||
# Overwrite any client-supplied value; Ktor must trust only this header.
|
||||
proxy_set_header X-OSG-mTLS-Verified "SUCCESS";
|
||||
proxy_set_header X-OSG-mTLS-Verified $ssl_client_verify;
|
||||
proxy_set_header Upgrade $http_upgrade;
|
||||
proxy_set_header Connection $connection_upgrade;
|
||||
proxy_hide_header Server;
|
||||
|
||||
@@ -470,9 +470,9 @@ WHERE version IS NOT NULL
|
||||
ORDER BY installed_rank;
|
||||
SQL
|
||||
)"
|
||||
EXPECTED_MIGRATIONS=$'1:1\n2:1\n3:1\n4:1\n5:1\n6:1\n7:1\n8:1\n9:1\n10:1\n11:1\n12:1'
|
||||
EXPECTED_MIGRATIONS=$'1:1\n2:1\n3:1\n4:1\n5:1\n6:1\n7:1\n8:1\n9:1\n10:1\n11:1\n12:1\n13:1\n14:1\n15:1\n16:1\n17:1'
|
||||
[[ "$MIGRATIONS" == "$EXPECTED_MIGRATIONS" ]] ||
|
||||
fail "Flyway history was not exactly successful V1-V12"
|
||||
fail "Flyway history was not exactly successful V1-V17"
|
||||
REFERRAL_REWARDS="$(
|
||||
mysql_root --batch --skip-column-names osg_account_smoke <<'SQL'
|
||||
SELECT CONCAT(inviter_reward_credits, ':', invitee_reward_credits)
|
||||
|
||||
@@ -17,6 +17,7 @@ GRANT SELECT ON osg_account_smoke.usage_records TO 'osg_smoke_runtime'@'%';
|
||||
GRANT SELECT ON osg_account_smoke.gateway_grants TO 'osg_smoke_runtime'@'%';
|
||||
GRANT SELECT ON osg_account_smoke.gateway_grant_scopes TO 'osg_smoke_runtime'@'%';
|
||||
GRANT SELECT ON osg_account_smoke.gateway_refresh_tokens TO 'osg_smoke_runtime'@'%';
|
||||
GRANT SELECT ON osg_account_smoke.gateway_complimentary_requests TO 'osg_smoke_runtime'@'%';
|
||||
GRANT SELECT ON osg_account_smoke.devicecheck_trial_claims TO 'osg_smoke_runtime'@'%';
|
||||
GRANT SELECT ON osg_account_smoke.app_attest_challenges TO 'osg_smoke_runtime'@'%';
|
||||
GRANT SELECT ON osg_account_smoke.app_attest_keys TO 'osg_smoke_runtime'@'%';
|
||||
@@ -48,6 +49,8 @@ GRANT INSERT ON osg_account_smoke.usage_records TO 'osg_smoke_runtime'@'%';
|
||||
GRANT INSERT, UPDATE ON osg_account_smoke.gateway_grants TO 'osg_smoke_runtime'@'%';
|
||||
GRANT INSERT ON osg_account_smoke.gateway_grant_scopes TO 'osg_smoke_runtime'@'%';
|
||||
GRANT INSERT, UPDATE ON osg_account_smoke.gateway_refresh_tokens TO 'osg_smoke_runtime'@'%';
|
||||
GRANT INSERT, UPDATE, DELETE ON osg_account_smoke.gateway_complimentary_requests
|
||||
TO 'osg_smoke_runtime'@'%';
|
||||
GRANT INSERT, UPDATE ON osg_account_smoke.devicecheck_trial_claims TO 'osg_smoke_runtime'@'%';
|
||||
GRANT INSERT, UPDATE ON osg_account_smoke.app_attest_challenges TO 'osg_smoke_runtime'@'%';
|
||||
GRANT INSERT, UPDATE ON osg_account_smoke.app_attest_keys TO 'osg_smoke_runtime'@'%';
|
||||
|
||||
+7
-1
@@ -132,10 +132,16 @@ docker compose logs --since=10m account-server
|
||||
2. 将三个 `server` 块作为站点配置;按 1Panel 实际证书路径调整 `ssl_certificate`。
|
||||
3. 示例 upstream 指向宿主机 `127.0.0.1:18080`。若 OpenResty 自身在容器中,则将其加入
|
||||
`account-backend`,并改为 `account-server:8080`。
|
||||
4. 配置明确对 `/admin`、`/internal`、`/v1/admin` 返回 404;不要新增绕过该规则的泛域名代理。
|
||||
4. `/internal` 始终返回 404;`/admin` 和 `/v1/admin` 是否要求客户端证书由
|
||||
`ADMIN_MTLS_REQUIRED` 控制,默认值为 `true`。
|
||||
5. API 示例按 IP 限制 20 请求/秒,邀请页限制 5 请求/秒,可基于真实流量谨慎调整。
|
||||
6. 代理统一支持 HTTP/1.1 Upgrade/Connection,因此当前 HTTP API 与后续 WebSocket 入口都可用。
|
||||
|
||||
临时关闭管理端 mTLS 时,在 1Panel/Compose 环境中显式设置
|
||||
`ADMIN_MTLS_REQUIRED=false` 并重启 `account-server`。此时管理端仍要求用户名、密码和 TOTP,
|
||||
且保留登录限流、失败锁定、同源校验、CSRF、会话 Cookie 与审计。恢复时将该值改回 `true`;
|
||||
不要删除客户端 CA、证书或轮换记录。
|
||||
|
||||
两个 AASA 地址由 Ktor 根据 `appleAppId` 模板输出,不需要复制静态文件。配置检查成功后再通过
|
||||
1Panel 重载 OpenResty:
|
||||
|
||||
|
||||
@@ -29,6 +29,7 @@ GRANT SELECT ON osg_account.usage_records TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT SELECT ON osg_account.gateway_grants TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT SELECT ON osg_account.gateway_grant_scopes TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT SELECT ON osg_account.gateway_refresh_tokens TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT SELECT ON osg_account.gateway_complimentary_requests TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT SELECT ON osg_account.devicecheck_trial_claims TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT SELECT ON osg_account.app_attest_challenges TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT SELECT ON osg_account.app_attest_keys TO 'osg_account_runtime'@'10.20.%';
|
||||
@@ -60,6 +61,8 @@ GRANT INSERT ON osg_account.usage_records TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT INSERT, UPDATE ON osg_account.gateway_grants TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT INSERT ON osg_account.gateway_grant_scopes TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT INSERT, UPDATE ON osg_account.gateway_refresh_tokens TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT INSERT, UPDATE, DELETE ON osg_account.gateway_complimentary_requests
|
||||
TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT INSERT, UPDATE ON osg_account.devicecheck_trial_claims TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT INSERT, UPDATE ON osg_account.app_attest_challenges TO 'osg_account_runtime'@'10.20.%';
|
||||
GRANT INSERT, UPDATE ON osg_account.app_attest_keys TO 'osg_account_runtime'@'10.20.%';
|
||||
|
||||
+11
-2
@@ -410,6 +410,9 @@ paths:
|
||||
and output-budget policy from `capability` plus optional `taskKind`. It
|
||||
never infers task type from `input` or `context`, and clients cannot
|
||||
supply provider parameters. Search and tools are currently disabled.
|
||||
An authenticated `oobe` purpose is accepted only for dictation polish.
|
||||
The first successful request per account is complimentary; later attempts
|
||||
fail without falling through to paid billing.
|
||||
parameters:
|
||||
- $ref: "#/components/parameters/RequestId"
|
||||
- name: capability
|
||||
@@ -529,7 +532,7 @@ paths:
|
||||
content:
|
||||
application/json:
|
||||
schema: { $ref: "#/components/schemas/AdminSessionState" }
|
||||
"404": { description: Verified administrator client certificate is absent }
|
||||
"404": { description: Verified administrator client certificate is absent while mTLS is required }
|
||||
/v1/admin/auth/login:
|
||||
post:
|
||||
security:
|
||||
@@ -887,7 +890,7 @@ components:
|
||||
bearerFormat: JWT
|
||||
adminMtls:
|
||||
type: mutualTLS
|
||||
description: Client certificate issued by the dedicated administrator CA.
|
||||
description: Client certificate issued by the dedicated administrator CA; required when ADMIN_MTLS_REQUIRED is true.
|
||||
adminSession:
|
||||
type: apiKey
|
||||
in: cookie
|
||||
@@ -1714,6 +1717,12 @@ components:
|
||||
type: ["string", "null"]
|
||||
enum: [hotword, null]
|
||||
description: Optional product entry point; hotword is accepted only for AI requests
|
||||
requestPurpose:
|
||||
type: ["string", "null"]
|
||||
enum: [oobe, null]
|
||||
description: |
|
||||
Optional server-audited billing purpose. `oobe` is valid only with
|
||||
`polish` and `dictation_polish`, and is complimentary once per account.
|
||||
CreateGatewayGrantRequest:
|
||||
type: object
|
||||
additionalProperties: false
|
||||
|
||||
@@ -73,6 +73,7 @@ import com.osglab.account.features.gateway.adapters.SessionIdentityAdapter
|
||||
import com.osglab.account.features.gateway.GatewaySettings
|
||||
import com.osglab.account.features.gateway.asr.AsrStreamingService
|
||||
import com.osglab.account.features.gateway.ports.CreditReservationPort
|
||||
import com.osglab.account.features.gateway.ports.ComplimentaryRequestPort
|
||||
import com.osglab.account.features.gateway.ports.GatewayAccessTokenPort
|
||||
import com.osglab.account.features.gateway.ports.GatewayGrantPort
|
||||
import com.osglab.account.features.gateway.ports.GatewayGrantRepository
|
||||
@@ -330,7 +331,7 @@ fun Application.module() {
|
||||
integrityRoutes(koin.get())
|
||||
}
|
||||
if (appConfig.admin.enabled) {
|
||||
adminWebRoutes()
|
||||
adminWebRoutes(appConfig)
|
||||
rateLimit(ADMIN_API_RATE_LIMIT) {
|
||||
adminApiRoutes(
|
||||
config = appConfig,
|
||||
@@ -507,6 +508,7 @@ fun accountServerModule(config: AppConfig): Module = module {
|
||||
single<GatewayGrantRepository> { get<ExposedGatewayRepository>() }
|
||||
single<GatewayGrantPort> { get<ExposedGatewayRepository>() }
|
||||
single<GatewayUsagePort> { get<ExposedGatewayRepository>() }
|
||||
single<ComplimentaryRequestPort> { get<ExposedGatewayRepository>() }
|
||||
single<AccountProvisioner> {
|
||||
AccountProvisioner { accountId, deviceCheckToken, displayName ->
|
||||
val granted = get<DeviceCheckTrialService>().claimAndGrant(accountId, deviceCheckToken)
|
||||
@@ -617,7 +619,7 @@ fun accountServerModule(config: AppConfig): Module = module {
|
||||
single {
|
||||
ProviderCatalog(configuredProviders(config, get()))
|
||||
}
|
||||
single { GatewayService(get(), get(), get(), get()) }
|
||||
single { GatewayService(get(), get(), get(), get(), get()) }
|
||||
single { GatewayReconciliationService(get(), get()) }
|
||||
single {
|
||||
InviteWebConfig(
|
||||
|
||||
@@ -161,6 +161,7 @@ data class AppConfig(
|
||||
}
|
||||
val admin = AdminConfig(
|
||||
enabled = adminEnabled,
|
||||
mtlsRequired = config.booleanOrDefault("app.admin.mtlsRequired", true),
|
||||
bootstrapEnabled = adminBootstrapEnabled,
|
||||
bootstrapOperatorId = config.optionalValue("app.admin.bootstrapOperatorId")
|
||||
?.let {
|
||||
@@ -434,6 +435,7 @@ data class IntegrityConfig(
|
||||
|
||||
data class AdminConfig(
|
||||
val enabled: Boolean = false,
|
||||
val mtlsRequired: Boolean = true,
|
||||
val bootstrapEnabled: Boolean = false,
|
||||
val bootstrapOperatorId: UUID? = null,
|
||||
val bootstrapUsername: String? = null,
|
||||
|
||||
@@ -34,6 +34,8 @@ import io.ktor.http.HttpHeaders
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.server.application.ApplicationCall
|
||||
import io.ktor.server.application.call
|
||||
import io.ktor.server.application.createRouteScopedPlugin
|
||||
import io.ktor.server.application.install
|
||||
import io.ktor.server.http.content.staticResources
|
||||
import io.ktor.server.plugins.BadRequestException
|
||||
import io.ktor.server.plugins.ratelimit.RateLimitName
|
||||
@@ -51,8 +53,13 @@ import java.time.Clock
|
||||
import java.time.Duration
|
||||
import java.util.UUID
|
||||
|
||||
fun Route.adminWebRoutes() {
|
||||
staticResources("/admin", "admin", index = "index.html")
|
||||
fun Route.adminWebRoutes(config: AppConfig) {
|
||||
route("/admin") {
|
||||
install(RequireVerifiedAdminEdge) {
|
||||
appConfig = config
|
||||
}
|
||||
staticResources("/", "admin", index = "index.html")
|
||||
}
|
||||
}
|
||||
|
||||
fun Route.adminApiRoutes(
|
||||
@@ -71,7 +78,7 @@ fun Route.adminApiRoutes(
|
||||
rateLimit(ADMIN_AUTH_RATE_LIMIT) {
|
||||
route("/auth") {
|
||||
get("/session") {
|
||||
if (!call.requireVerifiedAdminEdge()) return@get
|
||||
if (!call.requireVerifiedAdminEdge(config)) return@get
|
||||
val principal = call.currentPrincipal(sessionService)
|
||||
call.respond(
|
||||
AdminSessionResponse(
|
||||
@@ -83,7 +90,7 @@ fun Route.adminApiRoutes(
|
||||
}
|
||||
|
||||
post("/login") {
|
||||
if (!call.requireVerifiedAdminEdge() || !call.requireSameOrigin(config)) return@post
|
||||
if (!call.requireVerifiedAdminEdge(config) || !call.requireSameOrigin(config)) return@post
|
||||
val request = call.receive<AdminLoginRequest>()
|
||||
val password = request.password.toCharArray()
|
||||
val result = try {
|
||||
@@ -119,7 +126,7 @@ fun Route.adminApiRoutes(
|
||||
}
|
||||
|
||||
post("/logout") {
|
||||
if (!call.requireVerifiedAdminEdge() || !call.requireSameOrigin(config)) return@post
|
||||
if (!call.requireVerifiedAdminEdge(config) || !call.requireSameOrigin(config)) return@post
|
||||
val sessionToken = call.request.cookies[SESSION_COOKIE]
|
||||
val csrfToken = call.request.header(CSRF_HEADER)
|
||||
if (
|
||||
@@ -141,7 +148,7 @@ fun Route.adminApiRoutes(
|
||||
}
|
||||
|
||||
get("/overview") {
|
||||
if (call.requirePrincipal(sessionService) == null) return@get
|
||||
if (call.requirePrincipal(config, sessionService) == null) return@get
|
||||
val stats = statsService.getRange(call.request.queryParameters["range"], clock)
|
||||
?: run {
|
||||
call.respond(HttpStatusCode.BadRequest, AdminErrorResponse("VALIDATION_ERROR"))
|
||||
@@ -151,7 +158,7 @@ fun Route.adminApiRoutes(
|
||||
}
|
||||
|
||||
get("/referrals") {
|
||||
if (call.requirePrincipal(sessionService) == null) return@get
|
||||
if (call.requirePrincipal(config, sessionService) == null) return@get
|
||||
val stats = statsService.getRange(call.request.queryParameters["range"], clock)
|
||||
?: run {
|
||||
call.respond(HttpStatusCode.BadRequest, AdminErrorResponse("VALIDATION_ERROR"))
|
||||
@@ -161,7 +168,7 @@ fun Route.adminApiRoutes(
|
||||
}
|
||||
|
||||
get("/analytics") {
|
||||
if (call.requirePrincipal(sessionService) == null) return@get
|
||||
if (call.requirePrincipal(config, sessionService) == null) return@get
|
||||
val window = parseAdminStatsRange(call.request.queryParameters["range"], clock)
|
||||
?: run {
|
||||
call.respond(HttpStatusCode.BadRequest, AdminErrorResponse("VALIDATION_ERROR"))
|
||||
@@ -173,6 +180,7 @@ fun Route.adminApiRoutes(
|
||||
get("/users") {
|
||||
if (
|
||||
call.requireRole(
|
||||
config,
|
||||
sessionService,
|
||||
setOf(AdminRole.SUPER_ADMIN, AdminRole.SUPPORT),
|
||||
) == null
|
||||
@@ -206,6 +214,7 @@ fun Route.adminApiRoutes(
|
||||
get("/users/{userId}") {
|
||||
if (
|
||||
call.requireRole(
|
||||
config,
|
||||
sessionService,
|
||||
setOf(AdminRole.SUPER_ADMIN, AdminRole.SUPPORT),
|
||||
) == null
|
||||
@@ -221,6 +230,7 @@ fun Route.adminApiRoutes(
|
||||
get("/users/{userId}/ledger") {
|
||||
if (
|
||||
call.requireRole(
|
||||
config,
|
||||
sessionService,
|
||||
setOf(AdminRole.SUPER_ADMIN, AdminRole.SUPPORT),
|
||||
) == null
|
||||
@@ -252,6 +262,7 @@ fun Route.adminApiRoutes(
|
||||
get("/credits/ledger") {
|
||||
if (
|
||||
call.requireRole(
|
||||
config,
|
||||
sessionService,
|
||||
setOf(AdminRole.SUPER_ADMIN, AdminRole.SUPPORT),
|
||||
) == null
|
||||
@@ -325,7 +336,7 @@ fun Route.adminApiRoutes(
|
||||
}
|
||||
|
||||
get("/operators/summary") {
|
||||
val principal = call.requirePrincipal(sessionService) ?: return@get
|
||||
val principal = call.requirePrincipal(config, sessionService) ?: return@get
|
||||
try {
|
||||
val summary = operatorService.summary(principal)
|
||||
call.respond(
|
||||
@@ -341,7 +352,7 @@ fun Route.adminApiRoutes(
|
||||
}
|
||||
|
||||
get("/operators") {
|
||||
val principal = call.requirePrincipal(sessionService) ?: return@get
|
||||
val principal = call.requirePrincipal(config, sessionService) ?: return@get
|
||||
val limit = call.pageLimit(maximum = 100) ?: run {
|
||||
call.respond(HttpStatusCode.BadRequest, AdminErrorResponse("VALIDATION_ERROR"))
|
||||
return@get
|
||||
@@ -468,7 +479,7 @@ fun Route.adminApiRoutes(
|
||||
}
|
||||
|
||||
get("/audit") {
|
||||
val principal = call.requirePrincipal(sessionService) ?: return@get
|
||||
val principal = call.requirePrincipal(config, sessionService) ?: return@get
|
||||
val limit = call.pageLimit(maximum = 100) ?: run {
|
||||
call.respond(HttpStatusCode.BadRequest, AdminErrorResponse("VALIDATION_ERROR"))
|
||||
return@get
|
||||
@@ -528,9 +539,10 @@ private fun parseAdminStatsRange(range: String?, clock: Clock): Pair<java.time.I
|
||||
}
|
||||
|
||||
private suspend fun ApplicationCall.requirePrincipal(
|
||||
config: AppConfig,
|
||||
sessions: AdminSessionService,
|
||||
): AdminPrincipal? {
|
||||
if (!requireVerifiedAdminEdge()) return null
|
||||
if (!requireVerifiedAdminEdge(config)) return null
|
||||
val principal = currentPrincipal(sessions)
|
||||
if (principal == null) {
|
||||
respond(HttpStatusCode.Unauthorized, AdminErrorResponse("UNAUTHORIZED"))
|
||||
@@ -539,10 +551,11 @@ private suspend fun ApplicationCall.requirePrincipal(
|
||||
}
|
||||
|
||||
private suspend fun ApplicationCall.requireRole(
|
||||
config: AppConfig,
|
||||
sessions: AdminSessionService,
|
||||
allowedRoles: Set<AdminRole>,
|
||||
): AdminPrincipal? {
|
||||
val principal = requirePrincipal(sessions) ?: return null
|
||||
val principal = requirePrincipal(config, sessions) ?: return null
|
||||
if (principal.role !in allowedRoles) {
|
||||
respond(HttpStatusCode.Forbidden, AdminErrorResponse("INSUFFICIENT_PERMISSION"))
|
||||
return null
|
||||
@@ -554,7 +567,7 @@ private suspend fun ApplicationCall.requireMutationPrincipal(
|
||||
config: AppConfig,
|
||||
sessions: AdminSessionService,
|
||||
): AdminPrincipal? {
|
||||
if (!requireVerifiedAdminEdge() || !requireSameOrigin(config)) return null
|
||||
if (!requireVerifiedAdminEdge(config) || !requireSameOrigin(config)) return null
|
||||
val sessionToken = request.cookies[SESSION_COOKIE]
|
||||
val csrfToken = request.header(CSRF_HEADER)
|
||||
val principal = if (sessionToken != null && csrfToken != null) {
|
||||
@@ -568,12 +581,31 @@ private suspend fun ApplicationCall.requireMutationPrincipal(
|
||||
return principal
|
||||
}
|
||||
|
||||
private suspend fun ApplicationCall.requireVerifiedAdminEdge(): Boolean {
|
||||
if (request.header(MTLS_HEADER) == MTLS_VERIFIED) return true
|
||||
private suspend fun ApplicationCall.requireVerifiedAdminEdge(config: AppConfig): Boolean {
|
||||
if (isVerifiedAdminEdge(config)) return true
|
||||
respond(HttpStatusCode.NotFound)
|
||||
return false
|
||||
}
|
||||
|
||||
private fun ApplicationCall.isVerifiedAdminEdge(config: AppConfig): Boolean =
|
||||
!config.admin.mtlsRequired || request.header(MTLS_HEADER) == MTLS_VERIFIED
|
||||
|
||||
private class AdminEdgePluginConfig {
|
||||
lateinit var appConfig: AppConfig
|
||||
}
|
||||
|
||||
private val RequireVerifiedAdminEdge = createRouteScopedPlugin(
|
||||
name = "RequireVerifiedAdminEdge",
|
||||
createConfiguration = ::AdminEdgePluginConfig,
|
||||
) {
|
||||
val appConfig = pluginConfig.appConfig
|
||||
onCall { call ->
|
||||
if (!call.isVerifiedAdminEdge(appConfig)) {
|
||||
call.respond(HttpStatusCode.NotFound)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun ApplicationCall.requireSameOrigin(config: AppConfig): Boolean {
|
||||
if (request.header(HttpHeaders.Origin) == config.publicBaseUrl) return true
|
||||
respond(HttpStatusCode.Forbidden, AdminErrorResponse("ORIGIN_INVALID"))
|
||||
|
||||
@@ -25,6 +25,12 @@ enum class GatewayRequestSource {
|
||||
HOTWORD,
|
||||
}
|
||||
|
||||
@Serializable
|
||||
enum class GatewayRequestPurpose {
|
||||
@SerialName("oobe")
|
||||
OOBE,
|
||||
}
|
||||
|
||||
@Serializable
|
||||
enum class UsageMeter {
|
||||
@SerialName("llm_token")
|
||||
@@ -61,6 +67,7 @@ data class TextGatewayRequest(
|
||||
val stream: Boolean = false,
|
||||
val requestSource: GatewayRequestSource? = null,
|
||||
val taskKind: GatewayTaskKind? = null,
|
||||
val requestPurpose: GatewayRequestPurpose? = null,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
@@ -154,6 +161,8 @@ sealed interface ProviderRequest {
|
||||
val capability: GatewayCapability
|
||||
val requestSource: GatewayRequestSource?
|
||||
get() = null
|
||||
val requestPurpose: GatewayRequestPurpose?
|
||||
get() = null
|
||||
}
|
||||
|
||||
data class TextProviderRequest(
|
||||
@@ -166,6 +175,7 @@ data class TextProviderRequest(
|
||||
val temperature: Double,
|
||||
val stream: Boolean,
|
||||
override val requestSource: GatewayRequestSource? = null,
|
||||
override val requestPurpose: GatewayRequestPurpose? = null,
|
||||
) : ProviderRequest
|
||||
|
||||
data class AsrProviderRequest(
|
||||
|
||||
@@ -3,6 +3,7 @@ package com.osglab.account.features.gateway.ports
|
||||
import com.osglab.account.features.gateway.models.GatewayCapability
|
||||
import com.osglab.account.features.gateway.models.GatewayGrant
|
||||
import com.osglab.account.features.gateway.models.GatewayPrincipal
|
||||
import com.osglab.account.features.gateway.models.GatewayRequestPurpose
|
||||
import com.osglab.account.features.gateway.models.GatewayRequestSource
|
||||
import com.osglab.account.features.gateway.models.ProviderUsage
|
||||
import com.osglab.account.features.gateway.models.UsageMeter
|
||||
@@ -14,6 +15,30 @@ data class CreditReservation(
|
||||
val reservedUnits: Long,
|
||||
)
|
||||
|
||||
data class ComplimentaryRequestClaim(
|
||||
val accountId: String,
|
||||
val purpose: GatewayRequestPurpose,
|
||||
val capability: GatewayCapability,
|
||||
val requestId: String,
|
||||
)
|
||||
|
||||
/**
|
||||
* Atomically grants a bounded complimentary request. Implementations must
|
||||
* enforce one consumed claim per account, purpose, and capability.
|
||||
*/
|
||||
interface ComplimentaryRequestPort {
|
||||
suspend fun claim(
|
||||
accountId: String,
|
||||
purpose: GatewayRequestPurpose,
|
||||
capability: GatewayCapability,
|
||||
requestId: String,
|
||||
): ComplimentaryRequestClaim?
|
||||
|
||||
suspend fun consume(claim: ComplimentaryRequestClaim)
|
||||
|
||||
suspend fun release(claim: ComplimentaryRequestClaim)
|
||||
}
|
||||
|
||||
data class ProviderUsageEstimate(
|
||||
val meter: UsageMeter,
|
||||
val units: Long,
|
||||
@@ -137,10 +162,11 @@ interface GatewayGrantRepository : GatewayGrantPort {
|
||||
data class ProviderRequestMetadata(
|
||||
val requestId: String,
|
||||
val accountId: String,
|
||||
val reservationId: String,
|
||||
val reservationId: String?,
|
||||
val providerId: String,
|
||||
val capability: GatewayCapability,
|
||||
val requestSource: GatewayRequestSource?,
|
||||
val requestPurpose: GatewayRequestPurpose? = null,
|
||||
)
|
||||
|
||||
data class ProviderRefund(
|
||||
|
||||
+93
-1
@@ -3,8 +3,11 @@ package com.osglab.account.features.gateway.repositories
|
||||
import com.osglab.account.config.DatabaseFactory
|
||||
import com.osglab.account.features.gateway.models.GatewayCapability
|
||||
import com.osglab.account.features.gateway.models.GatewayGrant
|
||||
import com.osglab.account.features.gateway.models.GatewayRequestPurpose
|
||||
import com.osglab.account.features.gateway.models.ProviderUsage
|
||||
import com.osglab.account.features.gateway.models.UsageMeter
|
||||
import com.osglab.account.features.gateway.ports.ComplimentaryRequestClaim
|
||||
import com.osglab.account.features.gateway.ports.ComplimentaryRequestPort
|
||||
import com.osglab.account.features.gateway.ports.GatewayGrantRepository
|
||||
import com.osglab.account.features.gateway.ports.GatewayRefreshRotationResult
|
||||
import com.osglab.account.features.gateway.ports.GatewayRequestAlreadyClaimedException
|
||||
@@ -19,13 +22,16 @@ import org.jetbrains.exposed.v1.core.and
|
||||
import org.jetbrains.exposed.v1.core.eq
|
||||
import org.jetbrains.exposed.v1.core.greater
|
||||
import org.jetbrains.exposed.v1.core.isNull
|
||||
import org.jetbrains.exposed.v1.core.lessEq
|
||||
import org.jetbrains.exposed.v1.core.or
|
||||
import org.jetbrains.exposed.v1.javatime.timestamp
|
||||
import org.jetbrains.exposed.v1.jdbc.deleteWhere
|
||||
import org.jetbrains.exposed.v1.jdbc.insert
|
||||
import org.jetbrains.exposed.v1.jdbc.insertIgnore
|
||||
import org.jetbrains.exposed.v1.jdbc.selectAll
|
||||
import org.jetbrains.exposed.v1.jdbc.update
|
||||
import java.time.Clock
|
||||
import java.time.Duration
|
||||
|
||||
private object ProviderRequestsTable : Table("provider_requests") {
|
||||
val requestId = varchar("request_id", 64)
|
||||
@@ -34,6 +40,7 @@ private object ProviderRequestsTable : Table("provider_requests") {
|
||||
val providerId = varchar("provider_id", 64)
|
||||
val capability = varchar("capability", 32)
|
||||
val requestSource = varchar("request_source", 32).nullable()
|
||||
val requestPurpose = varchar("request_purpose", 32).nullable()
|
||||
val status = varchar("status", 24)
|
||||
val providerRequestId = varchar("provider_request_id", 128).nullable()
|
||||
val usageMeter = varchar("usage_meter", 32).nullable()
|
||||
@@ -47,6 +54,18 @@ private object ProviderRequestsTable : Table("provider_requests") {
|
||||
override val primaryKey = PrimaryKey(accountId, requestId)
|
||||
}
|
||||
|
||||
private object ComplimentaryRequestsTable : Table("gateway_complimentary_requests") {
|
||||
val accountId = varchar("account_id", 36)
|
||||
val purpose = varchar("purpose", 32)
|
||||
val capability = varchar("capability", 32)
|
||||
val requestId = varchar("request_id", 64)
|
||||
val status = varchar("status", 16)
|
||||
val expiresAt = timestamp("expires_at")
|
||||
val createdAt = timestamp("created_at")
|
||||
val updatedAt = timestamp("updated_at")
|
||||
override val primaryKey = PrimaryKey(accountId, purpose, capability)
|
||||
}
|
||||
|
||||
private object UsageRecordsTable : Table("usage_records") {
|
||||
val id = long("id").autoIncrement()
|
||||
val accountId = varchar("account_id", 36)
|
||||
@@ -91,7 +110,7 @@ private object GatewayRefreshTokensTable : Table("gateway_refresh_tokens") {
|
||||
class ExposedGatewayRepository(
|
||||
private val databaseFactory: DatabaseFactory,
|
||||
private val clock: Clock = Clock.systemUTC(),
|
||||
) : GatewayGrantRepository, GatewayUsagePort {
|
||||
) : GatewayGrantRepository, GatewayUsagePort, ComplimentaryRequestPort {
|
||||
override suspend fun isAllowed(accountId: String, capability: GatewayCapability): Boolean =
|
||||
databaseFactory.query {
|
||||
val now = clock.instant()
|
||||
@@ -272,6 +291,65 @@ class ExposedGatewayRepository(
|
||||
?.takeIf { it.scopes == scopes }
|
||||
}
|
||||
|
||||
override suspend fun claim(
|
||||
accountId: String,
|
||||
purpose: GatewayRequestPurpose,
|
||||
capability: GatewayCapability,
|
||||
requestId: String,
|
||||
): ComplimentaryRequestClaim? = databaseFactory.query {
|
||||
val now = clock.instant()
|
||||
val expiresAt = now.plus(COMPLIMENTARY_CLAIM_TTL)
|
||||
val inserted = ComplimentaryRequestsTable.insertIgnore {
|
||||
it[ComplimentaryRequestsTable.accountId] = accountId
|
||||
it[ComplimentaryRequestsTable.purpose] = purpose.name
|
||||
it[ComplimentaryRequestsTable.capability] = capability.name
|
||||
it[ComplimentaryRequestsTable.requestId] = requestId
|
||||
it[status] = COMPLIMENTARY_CLAIMED
|
||||
it[ComplimentaryRequestsTable.expiresAt] = expiresAt
|
||||
it[createdAt] = now
|
||||
it[updatedAt] = now
|
||||
}.insertedCount == 1
|
||||
val reclaimed = if (!inserted) {
|
||||
ComplimentaryRequestsTable.update({
|
||||
complimentaryKey(accountId, purpose, capability) and
|
||||
(ComplimentaryRequestsTable.status eq COMPLIMENTARY_CLAIMED) and
|
||||
(ComplimentaryRequestsTable.expiresAt lessEq now)
|
||||
}) {
|
||||
it[ComplimentaryRequestsTable.requestId] = requestId
|
||||
it[ComplimentaryRequestsTable.expiresAt] = expiresAt
|
||||
it[updatedAt] = now
|
||||
} == 1
|
||||
} else {
|
||||
false
|
||||
}
|
||||
if (!inserted && !reclaimed) return@query null
|
||||
ComplimentaryRequestClaim(accountId, purpose, capability, requestId)
|
||||
}
|
||||
|
||||
override suspend fun consume(claim: ComplimentaryRequestClaim) {
|
||||
databaseFactory.query {
|
||||
val changed = ComplimentaryRequestsTable.update({
|
||||
complimentaryKey(claim.accountId, claim.purpose, claim.capability) and
|
||||
(ComplimentaryRequestsTable.requestId eq claim.requestId) and
|
||||
(ComplimentaryRequestsTable.status eq COMPLIMENTARY_CLAIMED)
|
||||
}) {
|
||||
it[status] = COMPLIMENTARY_CONSUMED
|
||||
it[updatedAt] = clock.instant()
|
||||
}
|
||||
check(changed == 1) { "Complimentary request cannot be consumed" }
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun release(claim: ComplimentaryRequestClaim) {
|
||||
databaseFactory.query {
|
||||
ComplimentaryRequestsTable.deleteWhere {
|
||||
complimentaryKey(claim.accountId, claim.purpose, claim.capability) and
|
||||
(ComplimentaryRequestsTable.requestId eq claim.requestId) and
|
||||
(ComplimentaryRequestsTable.status eq COMPLIMENTARY_CLAIMED)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun claim(metadata: ProviderRequestMetadata) {
|
||||
databaseFactory.query {
|
||||
val inserted = ProviderRequestsTable.insertIgnore {
|
||||
@@ -281,6 +359,7 @@ class ExposedGatewayRepository(
|
||||
it[providerId] = metadata.providerId
|
||||
it[capability] = metadata.capability.name
|
||||
it[requestSource] = metadata.requestSource?.name
|
||||
it[requestPurpose] = metadata.requestPurpose?.name
|
||||
it[status] = ProviderRequestState.CLAIMED.name
|
||||
it[createdAt] = clock.instant()
|
||||
}.insertedCount == 1
|
||||
@@ -352,6 +431,7 @@ class ExposedGatewayRepository(
|
||||
val changed = ProviderRequestsTable.update({
|
||||
requestKey(accountId, requestId) and
|
||||
(
|
||||
(ProviderRequestsTable.status eq ProviderRequestState.STARTED.name) or
|
||||
(ProviderRequestsTable.status eq ProviderRequestState.SETTLEMENT_PENDING.name) or
|
||||
(ProviderRequestsTable.status eq ProviderRequestState.SETTLED.name)
|
||||
)
|
||||
@@ -482,6 +562,18 @@ private fun requestKey(accountId: String, requestId: String) =
|
||||
(ProviderRequestsTable.accountId eq accountId) and
|
||||
(ProviderRequestsTable.requestId eq requestId)
|
||||
|
||||
private fun complimentaryKey(
|
||||
accountId: String,
|
||||
purpose: GatewayRequestPurpose,
|
||||
capability: GatewayCapability,
|
||||
) = (ComplimentaryRequestsTable.accountId eq accountId) and
|
||||
(ComplimentaryRequestsTable.purpose eq purpose.name) and
|
||||
(ComplimentaryRequestsTable.capability eq capability.name)
|
||||
|
||||
private fun org.jetbrains.exposed.v1.core.ResultRow.requestState(): ProviderRequestState =
|
||||
runCatching { ProviderRequestState.valueOf(this[ProviderRequestsTable.status]) }
|
||||
.getOrDefault(ProviderRequestState.MANUAL_REVIEW)
|
||||
|
||||
private val COMPLIMENTARY_CLAIM_TTL: Duration = Duration.ofMinutes(15)
|
||||
private const val COMPLIMENTARY_CLAIMED = "CLAIMED"
|
||||
private const val COMPLIMENTARY_CONSUMED = "CONSUMED"
|
||||
|
||||
@@ -25,6 +25,7 @@ import com.osglab.account.features.gateway.ports.GatewayPrincipalResolver
|
||||
import com.osglab.account.features.gateway.ports.GatewayRequestAlreadyClaimedException
|
||||
import com.osglab.account.features.gateway.providers.UnsupportedGatewayCapabilityException
|
||||
import com.osglab.account.features.gateway.services.GatewayAccessDeniedException
|
||||
import com.osglab.account.features.gateway.services.ComplimentaryRequestUnavailableException
|
||||
import com.osglab.account.features.gateway.services.GatewayGrantService
|
||||
import com.osglab.account.features.gateway.services.GatewayRefreshTokenInvalidException
|
||||
import com.osglab.account.features.gateway.services.GatewayRefreshTokenReuseException
|
||||
@@ -274,6 +275,7 @@ fun Route.configureGatewayRoutes(
|
||||
temperature = body.temperature,
|
||||
stream = body.stream,
|
||||
requestSource = body.requestSource,
|
||||
requestPurpose = body.requestPurpose,
|
||||
)
|
||||
|
||||
if (body.stream) {
|
||||
@@ -435,6 +437,13 @@ private suspend fun ApplicationCall.respondGatewayFailure(
|
||||
requestId,
|
||||
)
|
||||
|
||||
is ComplimentaryRequestUnavailableException -> respondGatewayError(
|
||||
HttpStatusCode.Conflict,
|
||||
"oobe_already_used",
|
||||
"The complimentary OOBE request has already been used",
|
||||
requestId,
|
||||
)
|
||||
|
||||
is GatewayBodyTooLargeException -> respondGatewayError(
|
||||
HttpStatusCode.PayloadTooLarge,
|
||||
"request_too_large",
|
||||
|
||||
@@ -2,6 +2,8 @@ package com.osglab.account.features.gateway.services
|
||||
|
||||
import com.osglab.account.features.gateway.models.AsrProviderRequest
|
||||
import com.osglab.account.features.gateway.models.GatewayCapability
|
||||
import com.osglab.account.features.gateway.models.GatewayRequestPurpose
|
||||
import com.osglab.account.features.gateway.models.GatewayTaskKind
|
||||
import com.osglab.account.features.gateway.models.GatewaySubject
|
||||
import com.osglab.account.features.gateway.models.ProviderOutput
|
||||
import com.osglab.account.features.gateway.models.ProviderRequest
|
||||
@@ -10,6 +12,8 @@ import com.osglab.account.features.gateway.models.TextProviderRequest
|
||||
import com.osglab.account.features.gateway.models.UsageMeter
|
||||
import com.osglab.account.features.gateway.ports.CreditReservation
|
||||
import com.osglab.account.features.gateway.ports.CreditReservationPort
|
||||
import com.osglab.account.features.gateway.ports.ComplimentaryRequestClaim
|
||||
import com.osglab.account.features.gateway.ports.ComplimentaryRequestPort
|
||||
import com.osglab.account.features.gateway.ports.GatewayGrantPort
|
||||
import com.osglab.account.features.gateway.ports.GatewayRequestAlreadyClaimedException
|
||||
import com.osglab.account.features.gateway.ports.GatewayUsagePort
|
||||
@@ -28,6 +32,7 @@ class GatewayService(
|
||||
private val credits: CreditReservationPort,
|
||||
private val grants: GatewayGrantPort,
|
||||
private val usageRecords: GatewayUsagePort,
|
||||
private val complimentaryRequests: ComplimentaryRequestPort = NoComplimentaryRequests,
|
||||
private val usageEstimator: GatewayUsageEstimator = ConservativeGatewayUsageEstimator,
|
||||
private val llmProviderTimeoutMillis: Long = 120_000L,
|
||||
private val asrProviderTimeoutMillis: Long = 360_000L,
|
||||
@@ -61,28 +66,46 @@ class GatewayService(
|
||||
val provider = catalog.providerFor(request)
|
||||
val estimate = usageEstimator.estimate(request)
|
||||
validateEstimate(request, estimate)
|
||||
val reservation = credits.reserve(
|
||||
val complimentaryClaim = request.requestPurpose?.let { purpose ->
|
||||
validateComplimentaryRequest(request, purpose)
|
||||
complimentaryRequests.claim(
|
||||
accountId = subject.accountId,
|
||||
purpose = purpose,
|
||||
capability = request.capability,
|
||||
requestId = request.requestId,
|
||||
) ?: throw ComplimentaryRequestUnavailableException(purpose)
|
||||
}
|
||||
val reservation = if (complimentaryClaim == null) {
|
||||
credits.reserve(
|
||||
accountId = subject.accountId,
|
||||
estimate = estimate,
|
||||
requestId = request.requestId,
|
||||
)
|
||||
} else {
|
||||
null
|
||||
}
|
||||
try {
|
||||
usageRecords.claim(
|
||||
ProviderRequestMetadata(
|
||||
requestId = request.requestId,
|
||||
accountId = subject.accountId,
|
||||
reservationId = reservation.id,
|
||||
reservationId = reservation?.id,
|
||||
providerId = provider.descriptor.id,
|
||||
capability = request.capability,
|
||||
requestSource = request.requestSource,
|
||||
requestPurpose = request.requestPurpose,
|
||||
),
|
||||
)
|
||||
} catch (replay: GatewayRequestAlreadyClaimedException) {
|
||||
// The existing claim owns the reservation. Releasing it here would
|
||||
// refund an in-flight or completed request.
|
||||
// refund an in-flight or completed paid request. Complimentary
|
||||
// claims are newly acquired above and must not remain stranded.
|
||||
if (complimentaryClaim != null) {
|
||||
releaseAfterFailure(null, complimentaryClaim, replay)
|
||||
}
|
||||
throw replay
|
||||
} catch (failure: Throwable) {
|
||||
releaseAfterFailure(reservation, failure)
|
||||
releaseAfterFailure(reservation, complimentaryClaim, failure)
|
||||
throw failure
|
||||
}
|
||||
|
||||
@@ -93,12 +116,20 @@ class GatewayService(
|
||||
subject.accountId,
|
||||
request.requestId,
|
||||
reservation,
|
||||
complimentaryClaim,
|
||||
failure,
|
||||
)
|
||||
throw failure
|
||||
}
|
||||
|
||||
return PreparedGatewayRequest(subject, request, provider, estimate, reservation)
|
||||
return PreparedGatewayRequest(
|
||||
subject,
|
||||
request,
|
||||
provider,
|
||||
estimate,
|
||||
reservation,
|
||||
complimentaryClaim,
|
||||
)
|
||||
}
|
||||
|
||||
suspend fun executePrepared(
|
||||
@@ -141,11 +172,16 @@ class GatewayService(
|
||||
// Once upstream has completed, cancellation must not interrupt durable
|
||||
// metering. The reservation remains frozen if any settlement step fails.
|
||||
withContext(NonCancellable) {
|
||||
if (prepared.complimentaryClaim != null) {
|
||||
settleComplimentary(prepared, usage)
|
||||
return@withContext
|
||||
}
|
||||
val billableReservation = requireNotNull(reservation)
|
||||
val pendingRecorded = runCatching {
|
||||
usageRecords.markSettlementPending(subject.accountId, request.requestId, usage)
|
||||
}.isSuccess
|
||||
|
||||
val settled = runCatching { credits.settle(reservation.id, usage) }.isSuccess
|
||||
val settled = runCatching { credits.settle(billableReservation.id, usage) }.isSuccess
|
||||
if (settled && pendingRecorded) {
|
||||
// Metadata failure after a successful settlement must not turn a
|
||||
// successful provider response into a client-visible 502. The
|
||||
@@ -177,6 +213,7 @@ class GatewayService(
|
||||
prepared.subject.accountId,
|
||||
prepared.request.requestId,
|
||||
prepared.reservation,
|
||||
prepared.complimentaryClaim,
|
||||
failure,
|
||||
)
|
||||
}
|
||||
@@ -198,10 +235,15 @@ class GatewayService(
|
||||
private suspend fun releaseAndRecord(
|
||||
accountId: String,
|
||||
requestId: String,
|
||||
reservation: CreditReservation,
|
||||
reservation: CreditReservation?,
|
||||
complimentaryClaim: ComplimentaryRequestClaim?,
|
||||
failure: Throwable,
|
||||
): Unit = withContext(NonCancellable) {
|
||||
val released = runCatching { credits.release(reservation.id) }
|
||||
val released = if (complimentaryClaim != null) {
|
||||
runCatching { complimentaryRequests.release(complimentaryClaim) }
|
||||
} else {
|
||||
runCatching { credits.release(requireNotNull(reservation).id) }
|
||||
}
|
||||
if (released.isSuccess) {
|
||||
runCatching {
|
||||
usageRecords.markReleased(
|
||||
@@ -218,6 +260,30 @@ class GatewayService(
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun settleComplimentary(
|
||||
prepared: PreparedGatewayRequest,
|
||||
usage: ProviderUsage,
|
||||
) {
|
||||
val claim = requireNotNull(prepared.complimentaryClaim)
|
||||
val consumed = runCatching { complimentaryRequests.consume(claim) }.isSuccess
|
||||
val recorded = consumed && runCatching {
|
||||
usageRecords.markSucceeded(
|
||||
prepared.subject.accountId,
|
||||
prepared.request.requestId,
|
||||
usage,
|
||||
)
|
||||
}.isSuccess
|
||||
if (!consumed || !recorded) {
|
||||
runCatching {
|
||||
usageRecords.markManualReview(
|
||||
prepared.subject.accountId,
|
||||
prepared.request.requestId,
|
||||
if (consumed) "complimentary_usage_record_pending" else "complimentary_consume_pending",
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun validateUsage(usage: ProviderUsage, estimate: ProviderUsageEstimate) {
|
||||
if (usage.meter != estimate.meter) {
|
||||
throw GatewayUsagePolicyException("Provider usage meter differs from the reservation")
|
||||
@@ -272,11 +338,31 @@ class GatewayService(
|
||||
}
|
||||
}
|
||||
|
||||
private fun validateComplimentaryRequest(
|
||||
request: ProviderRequest,
|
||||
purpose: GatewayRequestPurpose,
|
||||
) {
|
||||
require(
|
||||
purpose == GatewayRequestPurpose.OOBE &&
|
||||
request is TextProviderRequest &&
|
||||
request.capability == GatewayCapability.POLISH &&
|
||||
request.executionPolicy.taskKind == GatewayTaskKind.DICTATION_POLISH,
|
||||
) {
|
||||
"OOBE is supported only for dictation polish"
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun releaseAfterFailure(
|
||||
reservation: CreditReservation,
|
||||
reservation: CreditReservation?,
|
||||
complimentaryClaim: ComplimentaryRequestClaim?,
|
||||
failure: Throwable,
|
||||
): Unit = withContext(NonCancellable) {
|
||||
runCatching { credits.release(reservation.id) }
|
||||
val released = if (complimentaryClaim != null) {
|
||||
runCatching { complimentaryRequests.release(complimentaryClaim) }
|
||||
} else {
|
||||
runCatching { credits.release(requireNotNull(reservation).id) }
|
||||
}
|
||||
released
|
||||
.onFailure(failure::addSuppressed)
|
||||
}
|
||||
|
||||
@@ -290,7 +376,8 @@ data class PreparedGatewayRequest(
|
||||
val request: ProviderRequest,
|
||||
val provider: GatewayProvider,
|
||||
val estimate: ProviderUsageEstimate,
|
||||
val reservation: CreditReservation,
|
||||
val reservation: CreditReservation?,
|
||||
val complimentaryClaim: ComplimentaryRequestClaim?,
|
||||
)
|
||||
|
||||
class GatewayReconciliationService(
|
||||
@@ -333,5 +420,22 @@ class GatewayRefundService(
|
||||
|
||||
class GatewayUsagePolicyException(message: String) : RuntimeException(message)
|
||||
|
||||
class ComplimentaryRequestUnavailableException(
|
||||
val purpose: GatewayRequestPurpose,
|
||||
) : RuntimeException("Complimentary ${purpose.name.lowercase()} request is unavailable")
|
||||
|
||||
class GatewayAccessDeniedException(capability: GatewayCapability) :
|
||||
RuntimeException("Gateway grant does not allow ${capability.name.lowercase()}")
|
||||
|
||||
private object NoComplimentaryRequests : ComplimentaryRequestPort {
|
||||
override suspend fun claim(
|
||||
accountId: String,
|
||||
purpose: GatewayRequestPurpose,
|
||||
capability: GatewayCapability,
|
||||
requestId: String,
|
||||
): ComplimentaryRequestClaim? = null
|
||||
|
||||
override suspend fun consume(claim: ComplimentaryRequestClaim) = Unit
|
||||
|
||||
override suspend fun release(claim: ComplimentaryRequestClaim) = Unit
|
||||
}
|
||||
|
||||
@@ -32,6 +32,7 @@ app:
|
||||
tombstoneRetentionDays: "$IDENTITY_TOMBSTONE_RETENTION_DAYS:365"
|
||||
admin:
|
||||
enabled: "$ADMIN_ENABLED:false"
|
||||
mtlsRequired: "$ADMIN_MTLS_REQUIRED:true"
|
||||
bootstrapEnabled: "$ADMIN_BOOTSTRAP_ENABLED:false"
|
||||
bootstrapOperatorId: "$ADMIN_BOOTSTRAP_OPERATOR_ID:"
|
||||
bootstrapUsername: "$ADMIN_BOOTSTRAP_USERNAME:"
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
ALTER TABLE provider_requests
|
||||
ADD COLUMN request_purpose VARCHAR(32) CHARACTER SET ascii COLLATE ascii_bin NULL
|
||||
AFTER request_source,
|
||||
ADD INDEX idx_provider_requests_purpose_created (request_purpose, created_at);
|
||||
|
||||
CREATE TABLE gateway_complimentary_requests (
|
||||
account_id VARCHAR(36) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
|
||||
purpose VARCHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
|
||||
capability VARCHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
|
||||
request_id VARCHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
|
||||
status VARCHAR(16) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
|
||||
expires_at TIMESTAMP(6) NOT NULL,
|
||||
created_at TIMESTAMP(6) NOT NULL,
|
||||
updated_at TIMESTAMP(6) NOT NULL,
|
||||
PRIMARY KEY (account_id, purpose, capability),
|
||||
INDEX idx_gateway_complimentary_expiry (status, expires_at),
|
||||
CONSTRAINT fk_gateway_complimentary_account
|
||||
FOREIGN KEY (account_id) REFERENCES accounts (id) ON DELETE CASCADE,
|
||||
CONSTRAINT chk_gateway_complimentary_status
|
||||
CHECK (status IN ('CLAIMED', 'CONSUMED'))
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
|
||||
@@ -17,6 +17,7 @@ class AppConfigTest : FunSpec({
|
||||
config.credits.signupTrial shouldBe 1_000
|
||||
config.credits.referralInviter shouldBe 1_000
|
||||
config.credits.referralInvitee shouldBe 1_000
|
||||
config.admin.mtlsRequired shouldBe true
|
||||
}
|
||||
|
||||
test("production rejects placeholder secrets") {
|
||||
@@ -68,6 +69,14 @@ class AppConfigTest : FunSpec({
|
||||
admin.bootstrapTotpSecretBase32 shouldBe null
|
||||
}
|
||||
|
||||
test("administrator mTLS can be disabled explicitly") {
|
||||
val config = validConfig("test").apply {
|
||||
put("app.admin.mtlsRequired", "false")
|
||||
}
|
||||
|
||||
AppConfig.from(config).admin.mtlsRequired shouldBe false
|
||||
}
|
||||
|
||||
test("admin bootstrap cannot be enabled while admin routes are disabled") {
|
||||
val config = validProductionConfig().apply {
|
||||
put("app.admin.enabled", "false")
|
||||
|
||||
@@ -169,6 +169,7 @@ class DeploymentConsistencyTest : FunSpec({
|
||||
val smokePrivileges = root.read("deploy/smoke/runtime-grants.sql")
|
||||
|
||||
compose shouldContain "ADMIN_BOOTSTRAP_ENABLED: \${ADMIN_BOOTSTRAP_ENABLED:-false}"
|
||||
compose shouldContain "ADMIN_MTLS_REQUIRED: \${ADMIN_MTLS_REQUIRED:-true}"
|
||||
privileges shouldContain "GRANT SELECT ON osg_account.admin_operators"
|
||||
privileges shouldContain "GRANT INSERT, UPDATE ON osg_account.admin_operators"
|
||||
privileges shouldContain "GRANT SELECT ON osg_account.admin_sessions"
|
||||
@@ -177,6 +178,9 @@ class DeploymentConsistencyTest : FunSpec({
|
||||
privileges shouldContain "GRANT INSERT ON osg_account.gateway_grant_scopes"
|
||||
privileges shouldContain "GRANT SELECT ON osg_account.gateway_refresh_tokens"
|
||||
privileges shouldContain "GRANT INSERT, UPDATE ON osg_account.gateway_refresh_tokens"
|
||||
privileges shouldContain "GRANT SELECT ON osg_account.gateway_complimentary_requests"
|
||||
privileges shouldContain
|
||||
"GRANT INSERT, UPDATE, DELETE ON osg_account.gateway_complimentary_requests"
|
||||
privileges shouldContain "GRANT SELECT ON osg_account.account_profiles"
|
||||
privileges shouldContain "GRANT INSERT, UPDATE ON osg_account.account_profiles"
|
||||
privileges shouldContain "GRANT INSERT ON osg_account.admin_audit_log"
|
||||
@@ -206,6 +210,8 @@ class DeploymentConsistencyTest : FunSpec({
|
||||
test("OpenResty proxies HTTP WebSocket invitations and both AASA paths safely") {
|
||||
val openResty = root.read("deploy/openresty-account.conf")
|
||||
|
||||
openResty shouldContain "proxy_set_header X-OSG-mTLS-Verified \$ssl_client_verify;"
|
||||
openResty shouldNotContain "proxy_set_header X-OSG-mTLS-Verified \"SUCCESS\";"
|
||||
openResty shouldContain "proxy_set_header Upgrade \$http_upgrade;"
|
||||
openResty shouldContain "proxy_set_header Connection \$connection_upgrade;"
|
||||
openResty shouldContain "location = /.well-known/apple-app-site-association"
|
||||
|
||||
@@ -37,7 +37,7 @@ class SmokeDeploymentTest : FunSpec({
|
||||
runner shouldContain "APPLE_JWKS_URL=http://127.0.0.1:9/"
|
||||
runner shouldContain "VOLCENGINE_ASR_ENDPOINT=ws://127.0.0.1:9/"
|
||||
runner shouldContain "DEEPSEEK_ENDPOINT=http://127.0.0.1:9/"
|
||||
runner shouldContain "Flyway history was not exactly successful V1-V12"
|
||||
runner shouldContain "Flyway history was not exactly successful V1-V17"
|
||||
runner shouldContain "default referral rewards were not 1000 credits for both accounts"
|
||||
runner shouldContain "active smaller credit rates did not match the V10 contract"
|
||||
runner shouldContain "first ledger page omitted nextCursor"
|
||||
@@ -51,7 +51,7 @@ class SmokeDeploymentTest : FunSpec({
|
||||
|
||||
test("runtime grants cover every migrated table without mutable history privileges") {
|
||||
val grants = root.read("deploy/smoke/runtime-grants.sql")
|
||||
val migrationTables = (1..16)
|
||||
val migrationTables = (1..17)
|
||||
.flatMap { version ->
|
||||
val migration = Files.list(root.resolve("src/main/resources/db/migration")).use { paths ->
|
||||
paths.filter { it.fileName.toString().startsWith("V${version}__") }
|
||||
@@ -69,6 +69,8 @@ class SmokeDeploymentTest : FunSpec({
|
||||
|
||||
grantedTables.sorted() shouldContainExactly migrationTables.sorted()
|
||||
grants shouldContain "GRANT INSERT, UPDATE, DELETE ON osg_account_smoke.admin_sessions"
|
||||
grants shouldContain
|
||||
"GRANT INSERT, UPDATE, DELETE ON osg_account_smoke.gateway_complimentary_requests"
|
||||
grants shouldNotContain "UPDATE ON osg_account_smoke.credit_ledger"
|
||||
grants shouldNotContain "DELETE ON osg_account_smoke.credit_ledger"
|
||||
grants shouldNotContain "UPDATE ON osg_account_smoke.admin_audit_log"
|
||||
|
||||
@@ -68,6 +68,16 @@ class AdminRoutesTest {
|
||||
response.bodyAsText() shouldContain """"authenticated":false"""
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `disabled mTLS allows anonymous session check without edge header`() = testApplication {
|
||||
application { installAdminTestRoutes(mtlsRequired = false) }
|
||||
|
||||
val response = client.get("/v1/admin/auth/session")
|
||||
|
||||
assertEquals(HttpStatusCode.OK, response.status)
|
||||
response.bodyAsText() shouldContain """"authenticated":false"""
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `authenticated session exposes role for client-side capability navigation`() = testApplication {
|
||||
application {
|
||||
@@ -103,9 +113,20 @@ class AdminRoutesTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `admin web resources are embedded`() = testApplication {
|
||||
fun `admin web resources are hidden when mTLS is required`() = testApplication {
|
||||
application {
|
||||
routing { adminWebRoutes() }
|
||||
routing { adminWebRoutes(adminTestConfig()) }
|
||||
}
|
||||
|
||||
val response = client.get("/admin/")
|
||||
|
||||
assertEquals(HttpStatusCode.NotFound, response.status)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `admin web resources are embedded when mTLS is disabled`() = testApplication {
|
||||
application {
|
||||
routing { adminWebRoutes(adminTestConfig(mtlsRequired = false)) }
|
||||
}
|
||||
|
||||
val response = client.get("/admin/")
|
||||
@@ -296,6 +317,7 @@ private fun io.ktor.server.application.Application.installAdminTestRoutes(
|
||||
operatorService: AdminOperatorService = mockk(relaxed = true),
|
||||
auditService: AdminAuditService = mockk(relaxed = true),
|
||||
usersService: AdminUsersService = mockk(relaxed = true),
|
||||
mtlsRequired: Boolean = true,
|
||||
) {
|
||||
install(ContentNegotiation) {
|
||||
json(Json { explicitNulls = false })
|
||||
@@ -305,11 +327,7 @@ private fun io.ktor.server.application.Application.installAdminTestRoutes(
|
||||
rateLimiter(limit = 20, refillPeriod = 1.minutes)
|
||||
}
|
||||
}
|
||||
val config = mockk<AppConfig> {
|
||||
every { publicBaseUrl } returns "https://account.osglab.com"
|
||||
every { isProduction } returns false
|
||||
every { admin } returns AdminConfig()
|
||||
}
|
||||
val config = adminTestConfig(mtlsRequired)
|
||||
routing {
|
||||
adminApiRoutes(
|
||||
config = config,
|
||||
@@ -325,6 +343,12 @@ private fun io.ktor.server.application.Application.installAdminTestRoutes(
|
||||
}
|
||||
}
|
||||
|
||||
private fun adminTestConfig(mtlsRequired: Boolean = true) = mockk<AppConfig> {
|
||||
every { publicBaseUrl } returns "https://account.osglab.com"
|
||||
every { isProduction } returns false
|
||||
every { admin } returns AdminConfig(mtlsRequired = mtlsRequired)
|
||||
}
|
||||
|
||||
private fun grantRouteFixture(
|
||||
failure: RuntimeException,
|
||||
): Pair<AdminSessionService, AdminGrantService> {
|
||||
|
||||
+135
@@ -0,0 +1,135 @@
|
||||
package com.osglab.account.features.gateway.repositories
|
||||
|
||||
import com.osglab.account.config.DatabaseConfig
|
||||
import com.osglab.account.config.DatabaseFactory
|
||||
import com.osglab.account.features.gateway.models.GatewayCapability
|
||||
import com.osglab.account.features.gateway.models.GatewayRequestPurpose
|
||||
import io.kotest.core.spec.style.FunSpec
|
||||
import io.kotest.matchers.shouldBe
|
||||
import io.kotest.matchers.shouldNotBe
|
||||
import java.sql.DriverManager
|
||||
import java.time.Clock
|
||||
import java.time.Instant
|
||||
import java.time.ZoneOffset
|
||||
import java.util.UUID
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.awaitAll
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import org.opentest4j.TestAbortedException
|
||||
import org.testcontainers.DockerClientFactory
|
||||
import org.testcontainers.containers.MySQLContainer
|
||||
|
||||
class GatewayComplimentaryRepositoryIntegrationTest : FunSpec({
|
||||
test("complimentary claims are atomic consumed once and releasable before success") {
|
||||
withGatewayDatabase { config, databaseFactory ->
|
||||
val now = Instant.parse("2026-08-20T08:00:00Z")
|
||||
val consumedAccount = UUID.randomUUID()
|
||||
val releasedAccount = UUID.randomUUID()
|
||||
insertAccounts(config, listOf(consumedAccount, releasedAccount), now)
|
||||
val repository = ExposedGatewayRepository(
|
||||
databaseFactory,
|
||||
Clock.fixed(now, ZoneOffset.UTC),
|
||||
)
|
||||
|
||||
val concurrentClaims = coroutineScope {
|
||||
(1..8).map { index ->
|
||||
async(Dispatchers.Default) {
|
||||
repository.claim(
|
||||
accountId = consumedAccount.toString(),
|
||||
purpose = GatewayRequestPurpose.OOBE,
|
||||
capability = GatewayCapability.POLISH,
|
||||
requestId = "concurrent-oobe-$index",
|
||||
)
|
||||
}
|
||||
}.awaitAll()
|
||||
}
|
||||
val winningClaim = concurrentClaims.filterNotNull().single()
|
||||
|
||||
repository.consume(winningClaim)
|
||||
repository.release(winningClaim)
|
||||
repository.claim(
|
||||
accountId = consumedAccount.toString(),
|
||||
purpose = GatewayRequestPurpose.OOBE,
|
||||
capability = GatewayCapability.POLISH,
|
||||
requestId = "consumed-replay",
|
||||
) shouldBe null
|
||||
|
||||
val releasedClaim = repository.claim(
|
||||
accountId = releasedAccount.toString(),
|
||||
purpose = GatewayRequestPurpose.OOBE,
|
||||
capability = GatewayCapability.POLISH,
|
||||
requestId = "released-first-attempt",
|
||||
)
|
||||
releasedClaim shouldNotBe null
|
||||
repository.release(requireNotNull(releasedClaim))
|
||||
repository.claim(
|
||||
accountId = releasedAccount.toString(),
|
||||
purpose = GatewayRequestPurpose.OOBE,
|
||||
capability = GatewayCapability.POLISH,
|
||||
requestId = "released-retry",
|
||||
) shouldNotBe null
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
private suspend fun withGatewayDatabase(
|
||||
block: suspend (DatabaseConfig, DatabaseFactory) -> Unit,
|
||||
) {
|
||||
val externalJdbcUrl = System.getenv("TEST_MYSQL_JDBC_URL")?.takeIf(String::isNotBlank)
|
||||
if (externalJdbcUrl == null && !DockerClientFactory.instance().isDockerAvailable) {
|
||||
throw TestAbortedException("Docker is unavailable; MySQL integration test skipped")
|
||||
}
|
||||
val mysql = if (externalJdbcUrl == null) {
|
||||
GatewayMySqlContainer("mysql:8.4")
|
||||
.withDatabaseName("osg_gateway_complimentary_test")
|
||||
.withUsername("test")
|
||||
.withPassword("test")
|
||||
.also(GatewayMySqlContainer::start)
|
||||
} else {
|
||||
null
|
||||
}
|
||||
val config = DatabaseConfig(
|
||||
jdbcUrl = externalJdbcUrl ?: requireNotNull(mysql).jdbcUrl,
|
||||
username = System.getenv("TEST_MYSQL_USER")?.takeIf(String::isNotBlank)
|
||||
?: mysql?.username
|
||||
?: "root",
|
||||
password = System.getenv("TEST_MYSQL_PASSWORD") ?: mysql?.password ?: "",
|
||||
maximumPoolSize = 8,
|
||||
)
|
||||
val databaseFactory = DatabaseFactory(config)
|
||||
try {
|
||||
databaseFactory.database
|
||||
block(config, databaseFactory)
|
||||
} finally {
|
||||
databaseFactory.close()
|
||||
mysql?.stop()
|
||||
}
|
||||
}
|
||||
|
||||
private fun insertAccounts(
|
||||
config: DatabaseConfig,
|
||||
accountIds: List<UUID>,
|
||||
now: Instant,
|
||||
) {
|
||||
DriverManager.getConnection(config.jdbcUrl, config.username, config.password).use { connection ->
|
||||
connection.prepareStatement(
|
||||
"""
|
||||
INSERT INTO accounts (id, apple_sub, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?)
|
||||
""".trimIndent(),
|
||||
).use { statement ->
|
||||
accountIds.forEach { accountId ->
|
||||
statement.setString(1, accountId.toString())
|
||||
statement.setString(2, "gateway-test-$accountId")
|
||||
statement.setTimestamp(3, java.sql.Timestamp.from(now))
|
||||
statement.setTimestamp(4, java.sql.Timestamp.from(now))
|
||||
statement.addBatch()
|
||||
}
|
||||
statement.executeBatch()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class GatewayMySqlContainer(image: String) :
|
||||
MySQLContainer<GatewayMySqlContainer>(image)
|
||||
+122
-1
@@ -2,6 +2,7 @@ package com.osglab.account.features.gateway.services
|
||||
|
||||
import com.osglab.account.features.gateway.models.GatewayCapability
|
||||
import com.osglab.account.features.gateway.models.GatewayPrincipal
|
||||
import com.osglab.account.features.gateway.models.GatewayRequestPurpose
|
||||
import com.osglab.account.features.gateway.models.GatewayRequestSource
|
||||
import com.osglab.account.features.gateway.models.ProviderDescriptor
|
||||
import com.osglab.account.features.gateway.models.ProviderOutput
|
||||
@@ -11,9 +12,13 @@ import com.osglab.account.features.gateway.models.TextProviderRequest
|
||||
import com.osglab.account.features.gateway.models.UsageMeter
|
||||
import com.osglab.account.features.gateway.ports.CreditMeterPort
|
||||
import com.osglab.account.features.gateway.ports.CreditReservation
|
||||
import com.osglab.account.features.gateway.ports.ComplimentaryRequestClaim
|
||||
import com.osglab.account.features.gateway.ports.ComplimentaryRequestPort
|
||||
import com.osglab.account.features.gateway.ports.GatewayRequestAlreadyClaimedException
|
||||
import com.osglab.account.features.gateway.ports.GatewayUsagePort
|
||||
import com.osglab.account.features.gateway.ports.PendingSettlement
|
||||
import com.osglab.account.features.gateway.ports.ProviderRequestMetadata
|
||||
import com.osglab.account.features.gateway.ports.ProviderRequestState
|
||||
import com.osglab.account.features.gateway.ports.ProviderUsageEstimate
|
||||
import com.osglab.account.features.gateway.providers.GatewayProvider
|
||||
import com.osglab.account.features.gateway.providers.ProviderCatalog
|
||||
@@ -80,6 +85,75 @@ class GatewayServiceBillingTest : StringSpec({
|
||||
usageRecords.lastClaim?.requestSource shouldBe GatewayRequestSource.HOTWORD
|
||||
}
|
||||
|
||||
"executes one OOBE dictation polish without reserving or settling credits" {
|
||||
val credits = FakeCredits()
|
||||
val complimentary = FakeComplimentaryRequests()
|
||||
val usageRecords = FakeUsageRecords()
|
||||
val service = service(
|
||||
credits = credits,
|
||||
provider = FakeProvider(capability = GatewayCapability.POLISH),
|
||||
usageRecords = usageRecords,
|
||||
complimentaryRequests = complimentary,
|
||||
)
|
||||
|
||||
service.execute(
|
||||
PRINCIPAL.copy(scopes = setOf(GatewayCapability.POLISH)),
|
||||
oobeRequest(),
|
||||
DISCARD_OUTPUT,
|
||||
)
|
||||
|
||||
credits.reserveCalls shouldBe 0
|
||||
credits.settled shouldBe emptyList()
|
||||
credits.released shouldBe emptyList()
|
||||
complimentary.consumed.size shouldBe 1
|
||||
complimentary.released shouldBe emptyList()
|
||||
usageRecords.lastClaim?.reservationId shouldBe null
|
||||
usageRecords.lastClaim?.requestPurpose shouldBe GatewayRequestPurpose.OOBE
|
||||
}
|
||||
|
||||
"rejects a second OOBE polish without falling through to paid billing" {
|
||||
val credits = FakeCredits()
|
||||
val complimentary = FakeComplimentaryRequests(available = false)
|
||||
val service = service(
|
||||
credits = credits,
|
||||
provider = FakeProvider(capability = GatewayCapability.POLISH),
|
||||
complimentaryRequests = complimentary,
|
||||
)
|
||||
|
||||
shouldThrow<ComplimentaryRequestUnavailableException> {
|
||||
service.execute(
|
||||
PRINCIPAL.copy(scopes = setOf(GatewayCapability.POLISH)),
|
||||
oobeRequest(),
|
||||
DISCARD_OUTPUT,
|
||||
)
|
||||
}
|
||||
|
||||
credits.reserveCalls shouldBe 0
|
||||
}
|
||||
|
||||
"releases a newly acquired OOBE claim when the request id is a replay" {
|
||||
val complimentary = FakeComplimentaryRequests()
|
||||
val service = service(
|
||||
credits = FakeCredits(),
|
||||
provider = FakeProvider(capability = GatewayCapability.POLISH),
|
||||
usageRecords = FakeUsageRecords(
|
||||
claimFailure = GatewayRequestAlreadyClaimedException(ProviderRequestState.RELEASED),
|
||||
),
|
||||
complimentaryRequests = complimentary,
|
||||
)
|
||||
|
||||
shouldThrow<GatewayRequestAlreadyClaimedException> {
|
||||
service.execute(
|
||||
PRINCIPAL.copy(scopes = setOf(GatewayCapability.POLISH)),
|
||||
oobeRequest(),
|
||||
DISCARD_OUTPUT,
|
||||
)
|
||||
}
|
||||
|
||||
complimentary.consumed shouldBe emptyList()
|
||||
complimentary.released.size shouldBe 1
|
||||
}
|
||||
|
||||
"uses one reservation when a buffered DeepSeek empty result succeeds on retry" {
|
||||
val credits = FakeCredits()
|
||||
var attempts = 0
|
||||
@@ -210,11 +284,13 @@ private fun service(
|
||||
credits: CreditMeterPort,
|
||||
provider: GatewayProvider,
|
||||
usageRecords: GatewayUsagePort = FakeUsageRecords(),
|
||||
complimentaryRequests: ComplimentaryRequestPort = FakeComplimentaryRequests(available = false),
|
||||
): GatewayService = GatewayService(
|
||||
catalog = ProviderCatalog(listOf(provider)),
|
||||
credits = credits,
|
||||
grants = { _, _ -> true },
|
||||
usageRecords = usageRecords,
|
||||
complimentaryRequests = complimentaryRequests,
|
||||
)
|
||||
|
||||
private fun request(requestSource: GatewayRequestSource? = null): TextProviderRequest {
|
||||
@@ -236,6 +312,25 @@ private fun request(requestSource: GatewayRequestSource? = null): TextProviderRe
|
||||
)
|
||||
}
|
||||
|
||||
private fun oobeRequest(): TextProviderRequest {
|
||||
val executionPolicy = GatewayTaskPolicyResolver().resolve(
|
||||
GatewayCapability.POLISH,
|
||||
requestedTaskKind = null,
|
||||
requestedMaxOutputTokens = 32,
|
||||
)
|
||||
return TextProviderRequest(
|
||||
requestId = "oobe-request-123",
|
||||
capability = GatewayCapability.POLISH,
|
||||
executionPolicy = executionPolicy,
|
||||
input = "hello",
|
||||
context = null,
|
||||
maxOutputTokens = executionPolicy.maxOutputTokens,
|
||||
temperature = 0.2,
|
||||
stream = false,
|
||||
requestPurpose = GatewayRequestPurpose.OOBE,
|
||||
)
|
||||
}
|
||||
|
||||
private class FakeCredits(
|
||||
private val failSettle: Boolean = false,
|
||||
private val idempotent: Boolean = false,
|
||||
@@ -286,10 +381,11 @@ private class FakeCredits(
|
||||
|
||||
private class FakeProvider(
|
||||
private val fail: Boolean = false,
|
||||
capability: GatewayCapability = GatewayCapability.AI,
|
||||
) : GatewayProvider {
|
||||
override val descriptor = ProviderDescriptor(
|
||||
id = "mock-deepseek",
|
||||
capabilities = setOf(GatewayCapability.AI),
|
||||
capabilities = setOf(capability),
|
||||
streaming = true,
|
||||
usageMeter = UsageMeter.LLM_TOKEN,
|
||||
)
|
||||
@@ -300,6 +396,29 @@ private class FakeProvider(
|
||||
}
|
||||
}
|
||||
|
||||
private class FakeComplimentaryRequests(
|
||||
private val available: Boolean = true,
|
||||
) : ComplimentaryRequestPort {
|
||||
val consumed = mutableListOf<ComplimentaryRequestClaim>()
|
||||
val released = mutableListOf<ComplimentaryRequestClaim>()
|
||||
|
||||
override suspend fun claim(
|
||||
accountId: String,
|
||||
purpose: GatewayRequestPurpose,
|
||||
capability: GatewayCapability,
|
||||
requestId: String,
|
||||
): ComplimentaryRequestClaim? =
|
||||
if (available) ComplimentaryRequestClaim(accountId, purpose, capability, requestId) else null
|
||||
|
||||
override suspend fun consume(claim: ComplimentaryRequestClaim) {
|
||||
consumed += claim
|
||||
}
|
||||
|
||||
override suspend fun release(claim: ComplimentaryRequestClaim) {
|
||||
released += claim
|
||||
}
|
||||
}
|
||||
|
||||
private class EmptyResultProvider : GatewayProvider {
|
||||
override val descriptor = ProviderDescriptor(
|
||||
id = "empty-provider",
|
||||
@@ -315,10 +434,12 @@ private class EmptyResultProvider : GatewayProvider {
|
||||
|
||||
private class FakeUsageRecords(
|
||||
private val pending: MutableList<PendingSettlement> = mutableListOf(),
|
||||
private val claimFailure: RuntimeException? = null,
|
||||
) : GatewayUsagePort {
|
||||
var lastClaim: ProviderRequestMetadata? = null
|
||||
|
||||
override suspend fun claim(metadata: ProviderRequestMetadata) {
|
||||
claimFailure?.let { throw it }
|
||||
lastClaim = metadata
|
||||
}
|
||||
override suspend fun markStarted(accountId: String, requestId: String) = Unit
|
||||
|
||||
Reference in New Issue
Block a user