fix(platform): harden sessions, scanner recovery, and CLI guidance (#801)

* fix(auth): recover from unreadable sessions

Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>

* fix(scanner): defer unavailable scan tasks safely

Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>

* fix(web): prefer the SkillHub CLI install command

Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>

* fix(auth): decode session cookies during recovery

Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>

* fix(runtime): address scanner and session review findings

Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>

* fix(scanner): defer all server-side outages

Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>

* docs(scanner): clarify deferred failure semantics

Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>

* test(scanner): cover recovery boundaries

Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>

* fix(scanner): register startup hook on router

Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>

* test(e2e): align install defaults and reuse auth session

Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>

---------

Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>
This commit is contained in:
XiaoSeS 2026-09-02 20:15:42 +08:00 • committed by GitHub
parent 2b831f31a9
commit fc7c59534a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
34 changed files with 764 additions and 87 deletions

View file

@ -6,6 +6,8 @@ services:
SKILL_SCANNER_LLM_API_KEY: ${SKILL_SCANNER_LLM_API_KEY:-}
SKILL_SCANNER_LLM_BASE_URL: ${SKILL_SCANNER_LLM_BASE_URL:-}
SKILL_SCANNER_LLM_MODEL: ${SKILL_SCANNER_LLM_MODEL:-}
SKILLHUB_SCANNER_MAX_CONCURRENT_SCANS: ${SKILLHUB_SCANNER_MAX_CONCURRENT_SCANS:-1}
SKILLHUB_SCANNER_HARD_TIMEOUT_SECONDS: ${SKILLHUB_SCANNER_HARD_TIMEOUT_SECONDS:-930}
healthcheck:
test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8000/health"]
interval: 10s
@ -87,6 +89,10 @@ services:
SKILLHUB_SECURITY_SCANNER_ENABLED: ${SKILLHUB_SECURITY_SCANNER_ENABLED:-true}
SKILLHUB_SECURITY_SCANNER_URL: http://skill-scanner:8000
SKILLHUB_SECURITY_SCANNER_MODE: upload
SKILLHUB_SECURITY_SCANNER_CONNECT_TIMEOUT: ${SKILLHUB_SECURITY_SCANNER_CONNECT_TIMEOUT:-5000}
SKILLHUB_SECURITY_SCANNER_READ_TIMEOUT: ${SKILLHUB_SECURITY_SCANNER_READ_TIMEOUT:-900000}
SKILLHUB_SECURITY_SCANNER_RETRY_MAX: ${SKILLHUB_SECURITY_SCANNER_RETRY_MAX:-3}
SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE: ${SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE:-PT16M}
SKILLHUB_BUILTIN_SKILLS_ENABLED: ${SKILLHUB_BUILTIN_SKILLS_ENABLED:-true}
SKILLHUB_AUTH_DIRECT_ENABLED: ${SKILLHUB_AUTH_DIRECT_ENABLED:-false}
SKILLHUB_TRACING_MODE: ${SKILLHUB_TRACING_MODE:-none}

View file

@ -160,6 +160,16 @@ spec:
configMapKeyRef:
name: skillhub-config
key: skill-scanner-mode
- name: SKILLHUB_SECURITY_SCANNER_READ_TIMEOUT
valueFrom:
configMapKeyRef:
name: skillhub-config
key: skill-scanner-read-timeout
- name: SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE
valueFrom:
configMapKeyRef:
name: skillhub-config
key: skill-scan-reclaim-min-idle
# Session
- name: SESSION_COOKIE_SECURE

View file

@ -35,6 +35,10 @@ data:
skill-scanner-enabled: "true"
skill-scanner-url: http://skillhub-scanner:8000
skill-scanner-mode: upload
skill-scanner-read-timeout: "900000"
skill-scan-reclaim-min-idle: PT16M
skill-scanner-max-concurrent-scans: "1"
skill-scanner-hard-timeout-seconds: "930"
# Bootstrap 管理员配置(非敏感)
bootstrap-admin-enabled: "true"

View file

@ -40,6 +40,16 @@ spec:
name: skillhub-secret
key: skill-scanner-llm-model
optional: true
- name: SKILLHUB_SCANNER_MAX_CONCURRENT_SCANS
valueFrom:
configMapKeyRef:
name: skillhub-config
key: skill-scanner-max-concurrent-scans
- name: SKILLHUB_SCANNER_HARD_TIMEOUT_SECONDS
valueFrom:
configMapKeyRef:
name: skillhub-config
key: skill-scanner-hard-timeout-seconds
readinessProbe:
httpGet:
path: /health

View file

@ -43,11 +43,12 @@ skillhub:
scan-path: /scan-upload
mode: local
connect-timeout-ms: 5000
read-timeout-ms: 300000
read-timeout-ms: 900000
retry-max-attempts: 3
stream:
key: skillhub:scan:requests
group: skillhub-scanners
reclaim-min-idle: PT16M
```
Important environment variables:
@ -55,16 +56,23 @@ Important environment variables:
- `SKILLHUB_SECURITY_SCANNER_ENABLED`
- `SKILLHUB_SECURITY_SCANNER_URL`
- `SKILLHUB_SECURITY_SCANNER_MODE`
- `SKILLHUB_SECURITY_SCANNER_READ_TIMEOUT`
- `SKILLHUB_SCAN_STREAM_KEY`
- `SKILLHUB_SCAN_STREAM_GROUP`
- `SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE`
Scanner-side optional environment variables:
- `SKILL_SCANNER_LLM_API_KEY`
- `SKILL_SCANNER_LLM_BASE_URL`
- `SKILL_SCANNER_LLM_MODEL`
- `SKILLHUB_SCANNER_MAX_CONCURRENT_SCANS` (default `1`)
- `SKILLHUB_SCANNER_HARD_TIMEOUT_SECONDS` (default `930`)
If the LLM variables are absent, the scanner should still run with non-LLM analyzers.
The default timeout ordering is server read timeout (900 seconds), scanner hard timeout
(930 seconds), then pending-message reclaim (960 seconds). A hard timeout exits the scanner process
with status `124`; Compose or Kubernetes restarts it and the Redis pending task is retried.
## Kubernetes Notes

View file

@ -128,22 +128,25 @@ In the browser, you can add the `X-Mock-User-Id` header via a browser extension
## Install the CLI Tool
SkillHub is compatible with the OpenClaw CLI. You can use the `npx clawhub` command to manage skill packages:
Use the first-party SkillHub CLI for skill package management:
```bash
# Configure the SkillHub registry URL
export CLAWHUB_REGISTRY=http://localhost:8080
# Install the CLI and configure the SkillHub registry URL
npm install -g @astron-team/skillhub
export SKILLHUB_REGISTRY=http://localhost:8080
# Search for skill packages
npx clawhub search email
skillhub search email
# Install a skill package
npx clawhub install my-skill
skillhub install my-skill
# Publish a skill package
npx clawhub publish ./my-skill
skillhub publish ./my-skill --namespace my-team
```
Existing ClawHub workflows can still use the compatibility layer for search and install. Prefer the SkillHub CLI for new workflows and publishing.
## Publish Your First Skill Package
### Publish via CLI (Recommended)
@ -164,13 +167,10 @@ my-skill/
```bash
# Configure the registry
export CLAWHUB_REGISTRY=http://localhost:8080
# Publish to the default namespace
npx clawhub publish ./my-skill
export SKILLHUB_REGISTRY=http://localhost:8080
# Publish to a specific namespace
npx clawhub publish ./my-skill --namespace my-team
skillhub publish ./my-skill --namespace my-team
```
3. **Wait for security scanning**
@ -201,13 +201,13 @@ Administrators will receive a notification and the skill package will be officia
```bash
# Search for skill packages
npx clawhub search pdf
skillhub search pdf
# Install a skill package
npx clawhub install pdf-parser
skillhub install pdf-parser
# Install a skill package from a specific namespace
npx clawhub install my-team--pdf-parser
skillhub install pdf-parser --namespace my-team
```
### Using the Web UI

View file

@ -128,22 +128,25 @@ curl -H "X-Mock-User-Id: local-admin" http://localhost:8080/api/v1/auth/me
## 安装 CLI 工具
SkillHub 兼容 OpenClaw CLI,可以使用 `npx clawhub` 命令管理技能包:
推荐使用第一方 SkillHub CLI 管理技能包:
```bash
# 配置 SkillHub 注册中心地址
export CLAWHUB_REGISTRY=http://localhost:8080
# 安装并配置 SkillHub 注册中心地址
npm install -g @astron-team/skillhub
export SKILLHUB_REGISTRY=http://localhost:8080
# 搜索技能包
npx clawhub search email
skillhub search email
# 安装技能包
npx clawhub install my-skill
skillhub install my-skill
# 发布技能包
npx clawhub publish ./my-skill
skillhub publish ./my-skill --namespace my-team
```
已有 ClawHub 工作流仍可通过兼容层进行搜索和安装;新流程及发布操作优先使用 SkillHub CLI。
## 发布第一个技能包
### 使用 CLI 工具发布(推荐)
@ -164,13 +167,10 @@ my-skill/
```bash
# 配置注册中心
export CLAWHUB_REGISTRY=http://localhost:8080
# 发布到默认命名空间
npx clawhub publish ./my-skill
export SKILLHUB_REGISTRY=http://localhost:8080
# 发布到指定命名空间
npx clawhub publish ./my-skill --namespace my-team
skillhub publish ./my-skill --namespace my-team
```
3. **等待安全扫描**
@ -201,13 +201,13 @@ npx clawhub publish ./my-skill --namespace my-team
```bash
# 搜索技能包
npx clawhub search pdf
skillhub search pdf
# 安装技能包
npx clawhub install pdf-parser
skillhub install pdf-parser
# 安装指定命名空间的技能包
npx clawhub install my-team--pdf-parser
skillhub install pdf-parser --namespace my-team
```
### 使用 Web UI

View file

@ -5,6 +5,7 @@ ARG SKILL_SCANNER_VERSION=1.0.2
WORKDIR /app
COPY backports/apply_1_0_2_llm_base_url_backport.py /tmp/apply_1_0_2_llm_base_url_backport.py
COPY skillhub_scanner_app.py /app/skillhub_scanner_app.py
RUN pip install --no-cache-dir \
"cisco-ai-skill-scanner==${SKILL_SCANNER_VERSION}" \
@ -23,4 +24,4 @@ EXPOSE 8000
HEALTHCHECK --interval=10s --timeout=3s \
CMD wget -qO- http://127.0.0.1:8000/health || exit 1
CMD ["skill-scanner-api", "--host", "0.0.0.0", "--port", "8000"]
CMD ["uvicorn", "skillhub_scanner_app:app", "--host", "0.0.0.0", "--port", "8000"]

View file

@ -21,6 +21,7 @@ skillhub:
enabled: ${SKILLHUB_SECURITY_SCANNER_ENABLED:true}
base-url: ${SKILLHUB_SECURITY_SCANNER_URL:http://localhost:8000}
mode: ${SKILLHUB_SECURITY_SCANNER_MODE:local}
read-timeout-ms: ${SKILLHUB_SECURITY_SCANNER_READ_TIMEOUT:900000}
# 分析器配置
analyzers:
@ -39,6 +40,10 @@ skillhub:
preset: ${SKILLHUB_SCANNER_POLICY_PRESET:balanced}
custom-policy-path: ${SKILLHUB_SCANNER_CUSTOM_POLICY_PATH:}
fail-on-severity: ${SKILLHUB_SCANNER_FAIL_ON_SEVERITY:high}
stream:
# Keep this greater than scanner.read-timeout-ms.
reclaim-min-idle: ${SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE:PT16M}
```
## 配置项详解
@ -110,6 +115,15 @@ mode: upload
mode: upload
```
#### 扫描超时与并发
- `SKILLHUB_SECURITY_SCANNER_READ_TIMEOUT`:服务端等待单次扫描的毫秒数,默认 15 分钟。
- `SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE`:未确认任务允许被恢复的等待时间,应大于扫描超时,默认 16 分钟。
- `SKILLHUB_SCANNER_MAX_CONCURRENT_SCANS`:Scanner 容器内的最大并发扫描数,默认 `1`;超出的请求返回 HTTP 503,由待处理消息稍后重试。
- `SKILLHUB_SCANNER_HARD_TIMEOUT_SECONDS`:Scanner 单次工作的硬上限,默认 930 秒。超时后进程以状态码 `124` 退出,由 Compose/Kubernetes 重启;默认关系为服务端等待 900 秒 < Scanner 硬上限 930 秒 < Redis 恢复等待 960 秒。进程退出会使同容器内其他扫描稍后重试,因此建议保持默认并发数 `1`。
Scanner 超时或暂时不可用(包括 429、5xx)时,版本不会进入 `SCAN_FAILED`。任务保留为待处理,版本保持 `SCANNING`,待 Scanner 恢复后自动继续。确定性的 4xx 包校验错误仍采用有限次数重试,最终可以进入 `SCAN_FAILED`。某个包若稳定触发 Scanner 内部 500,会保持 `SCANNING` 等待 Scanner 修复,而不会被误判为包本身不合格。
---
### 2. 分析器配置

View file

@ -0,0 +1,87 @@
"""Runtime safeguards around the upstream Cisco Skill Scanner ASGI application."""
import asyncio
import logging
import os
import shutil
import tempfile
from pathlib import Path
from typing import NoReturn
from fastapi import Request
from fastapi.responses import JSONResponse
from skill_scanner.api.api import app
_MAX_CONCURRENT_SCANS = max(1, int(os.getenv("SKILLHUB_SCANNER_MAX_CONCURRENT_SCANS", "1")))
_HARD_TIMEOUT_SECONDS = max(1, int(os.getenv("SKILLHUB_SCANNER_HARD_TIMEOUT_SECONDS", "930")))
_active_scans = 0
_active_scans_guard = asyncio.Lock()
_SCAN_PATHS = {"/scan", "/scan-upload"}
_log = logging.getLogger(__name__)
def _cleanup_stale_scan_directories(temp_root: Path | None = None) -> None:
"""Remove incomplete upstream extraction directories left by a process restart."""
root = temp_root or Path(tempfile.gettempdir())
for candidate in root.glob("skill_scanner_*"):
if not candidate.is_dir():
continue
try:
shutil.rmtree(candidate)
except OSError as error:
_log.warning("Could not remove stale scanner directory %s: %s", candidate, error)
def _restart_after_hard_timeout(request_path: str) -> NoReturn:
"""Terminate the single-scan worker so the container runtime can recover it."""
_log.critical(
"Security scan exceeded the %s second hard timeout: path=%s; restarting scanner",
_HARD_TIMEOUT_SECONDS,
request_path,
)
logging.shutdown()
os._exit(124)
async def _await_scan_until(scan_task: asyncio.Task, deadline: float, request_path: str):
remaining = deadline - asyncio.get_running_loop().time()
if remaining <= 0:
_restart_after_hard_timeout(request_path)
try:
return await asyncio.wait_for(asyncio.shield(scan_task), timeout=remaining)
except asyncio.TimeoutError:
_restart_after_hard_timeout(request_path)
app.router.add_event_handler("startup", _cleanup_stale_scan_directories)
@app.middleware("http")
async def limit_concurrent_scans(request: Request, call_next):
"""Reject excess scan work so timed-out client retries cannot multiply memory use."""
global _active_scans
if request.method != "POST" or request.url.path not in _SCAN_PATHS:
return await call_next(request)
async with _active_scans_guard:
if _active_scans >= _MAX_CONCURRENT_SCANS:
return JSONResponse(
status_code=503,
content={"detail": "Scanner is busy; retry later"},
headers={"Retry-After": "30"},
)
_active_scans += 1
scan_task = asyncio.create_task(call_next(request))
deadline = asyncio.get_running_loop().time() + _HARD_TIMEOUT_SECONDS
try:
# Keep the capacity slot until upstream work really ends, even if the HTTP client
# disconnects while the scanner's worker thread is still running.
return await _await_scan_until(scan_task, deadline, request.url.path)
except asyncio.CancelledError:
await _await_scan_until(scan_task, deadline, request.url.path)
raise
finally:
async with _active_scans_guard:
_active_scans -= 1

View file

@ -0,0 +1,131 @@
import asyncio
import importlib.util
import sys
import tempfile
import types
import unittest
from pathlib import Path
from unittest.mock import patch
class _FakeRouter:
def __init__(self):
self.handlers = []
def add_event_handler(self, _event, _handler):
self.handlers.append((_event, _handler))
class _FakeApp:
def __init__(self):
self.router = _FakeRouter()
def middleware(self, _kind):
return lambda function: function
class _FakeResponse:
def __init__(self, status_code, content, headers):
self.status_code = status_code
self.content = content
self.headers = headers
class _Request:
method = "POST"
url = types.SimpleNamespace(path="/scan-upload")
def _load_module():
fastapi = types.ModuleType("fastapi")
fastapi.Request = object
responses = types.ModuleType("fastapi.responses")
responses.JSONResponse = _FakeResponse
api = types.ModuleType("skill_scanner.api.api")
api.app = _FakeApp()
stubs = {
"fastapi": fastapi,
"fastapi.responses": responses,
"skill_scanner": types.ModuleType("skill_scanner"),
"skill_scanner.api": types.ModuleType("skill_scanner.api"),
"skill_scanner.api.api": api,
}
with patch.dict(sys.modules, stubs):
module_path = Path(__file__).parents[1] / "skillhub_scanner_app.py"
spec = importlib.util.spec_from_file_location("skillhub_scanner_app_under_test", module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class SkillHubScannerAppTest(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self):
self.module = _load_module()
async def test_excess_scan_is_rejected(self):
self.module._active_scans = 1
response = await self.module.limit_concurrent_scans(_Request(), lambda _request: None)
self.assertEqual(503, response.status_code)
self.assertEqual("30", response.headers["Retry-After"])
async def test_client_disconnect_keeps_slot_until_scan_finishes(self):
release = asyncio.Event()
async def scan(_request):
await release.wait()
return "done"
request_task = asyncio.create_task(self.module.limit_concurrent_scans(_Request(), scan))
await asyncio.sleep(0)
request_task.cancel()
await asyncio.sleep(0)
self.assertEqual(1, self.module._active_scans)
response = await self.module.limit_concurrent_scans(_Request(), scan)
self.assertEqual(503, response.status_code)
release.set()
with self.assertRaises(asyncio.CancelledError):
await request_task
self.assertEqual(0, self.module._active_scans)
async def test_hard_timeout_requests_process_restart(self):
self.module._HARD_TIMEOUT_SECONDS = 0.01
async def stuck_scan(_request):
await asyncio.Event().wait()
with patch.object(
self.module,
"_restart_after_hard_timeout",
side_effect=RuntimeError("restart requested")) as restart:
with self.assertRaisesRegex(RuntimeError, "restart requested"):
await self.module.limit_concurrent_scans(_Request(), stuck_scan)
restart.assert_called_once_with("/scan-upload")
self.assertEqual(0, self.module._active_scans)
async def test_startup_cleanup_removes_only_scanner_directories(self):
with tempfile.TemporaryDirectory() as temp_root:
root = Path(temp_root)
stale = root / "skill_scanner_abcd"
unrelated = root / "skillhub-data"
stale.mkdir()
unrelated.mkdir()
self.module._cleanup_stale_scan_directories(root)
self.assertFalse(stale.exists())
self.assertTrue(unrelated.exists())
async def test_startup_cleanup_is_registered_on_the_upstream_router(self):
self.assertEqual(
[("startup", self.module._cleanup_stale_scan_directories)],
self.module.app.router.handlers,
)
if __name__ == "__main__":
unittest.main()

View file

@ -28,7 +28,7 @@ public class RedisStreamConfig {
@Value("${skillhub.security.stream.reclaim-enabled:true}")
private boolean reclaimEnabled;
@Value("${skillhub.security.stream.reclaim-min-idle:PT2M}")
@Value("${skillhub.security.stream.reclaim-min-idle:PT16M}")
private Duration reclaimMinIdle;
@Value("${skillhub.security.stream.reclaim-batch-size:20}")
@ -37,6 +37,9 @@ public class RedisStreamConfig {
@Value("${skillhub.security.stream.reclaim-interval:PT30S}")
private Duration reclaimInterval;
@Value("${skillhub.security.scanner.retry-max-attempts:3}")
private int maxRetryAttempts;
@Bean
public RedissonScanTaskProducer redisScanTaskProducer(
RedissonClient redissonClient,
@ -66,6 +69,7 @@ public class RedisStreamConfig {
reclaimMinIdle,
reclaimBatchSize,
reclaimInterval,
maxRetryAttempts,
messageObservationSupport
);
}

View file

@ -0,0 +1,26 @@
package com.iflytek.skillhub.config;
import com.iflytek.skillhub.auth.session.CorruptSessionRemover;
import java.util.List;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.StringRedisTemplate;
/**
* Connects the authentication module's recovery boundary to Spring Session storage.
*/
@Configuration
public class SessionRecoveryConfig {
@Bean
CorruptSessionRemover corruptSessionRemover(
StringRedisTemplate redisTemplate,
@Value("${spring.session.redis.namespace:spring:session}") String namespace) {
String keyPrefix = namespace.endsWith(":") ? namespace : namespace + ":";
return sessionId -> redisTemplate.delete(List.of(
keyPrefix + "sessions:" + sessionId,
keyPrefix + "sessions:expires:" + sessionId
));
}
}

View file

@ -12,7 +12,7 @@ public class SkillScannerProperties {
private String healthPath = "/health";
private String scanPath = "/scan-upload";
private int connectTimeoutMs = 5000;
private int readTimeoutMs = 300000;
private int readTimeoutMs = 900000;
private int retryMaxAttempts = 3;
private String mode = "local";
private Analyzers analyzers = new Analyzers();

View file

@ -28,8 +28,8 @@ public abstract class AbstractStreamConsumer<T> {
protected final Logger log = LoggerFactory.getLogger(getClass());
private static final String FIELD_RETRY_COUNT = "retryCount";
private static final int MAX_RETRY_COUNT = 3;
private static final int READ_BATCH_SIZE = 10;
private static final int DEFAULT_MAX_RETRY_COUNT = 3;
private static final int DEFAULT_READ_BATCH_SIZE = 10;
private static final Duration POLL_TIMEOUT = Duration.ofSeconds(2);
private final RedissonClient redissonClient;
@ -170,7 +170,7 @@ public abstract class AbstractStreamConsumer<T> {
groupName,
consumerName,
StreamReadGroupArgs.neverDelivered()
.count(READ_BATCH_SIZE)
.count(readBatchSize())
.timeout(POLL_TIMEOUT)
);
processMessages(messages);
@ -237,13 +237,17 @@ public abstract class AbstractStreamConsumer<T> {
acknowledge(messageId);
} catch (Exception e) {
messageObservationSupport.recordCurrentError(e);
handleFailure(payload, retryCount, e);
acknowledge(messageId);
if (shouldDeferFailure(payload, e)) {
markDeferred(payload, e);
} else {
handleFailure(payload, retryCount, e);
acknowledge(messageId);
}
}
}
private void handleFailure(T payload, int retryCount, Exception e) {
if (retryCount < MAX_RETRY_COUNT) {
if (retryCount < maxRetryCount()) {
// Retry publication remains inside the current consumer scope, so the new producer
// span and message carrier continue the original trace.
retryMessage(payload, retryCount + 1);
@ -274,7 +278,28 @@ public abstract class AbstractStreamConsumer<T> {
}
protected void acknowledge(StreamMessageId messageId) {
stream().ack(groupName, messageId);
if (stream().ack(groupName, messageId) > 0) {
// This stream has one consumer group. Removing acknowledged entries prevents
// completed scan tasks from growing the Redis stream without bound.
stream().remove(messageId);
}
}
protected int readBatchSize() {
return DEFAULT_READ_BATCH_SIZE;
}
protected int maxRetryCount() {
return DEFAULT_MAX_RETRY_COUNT;
}
protected boolean shouldDeferFailure(T payload, Exception error) {
return false;
}
protected void markDeferred(T payload, Exception error) {
log.warn("Deferring {} task without acknowledging its stream entry: {}",
taskDisplayName(), payloadIdentifier(payload), error);
}
protected final RStream<String, String> stream() {

View file

@ -11,6 +11,7 @@ import com.iflytek.skillhub.domain.skill.SkillVersionRepository;
import com.iflytek.skillhub.domain.skill.SkillVersionStatus;
import com.iflytek.skillhub.observability.MessageObservationSupport;
import com.iflytek.skillhub.storage.ObjectStorageService;
import com.iflytek.skillhub.infra.scanner.SecurityScanException;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
@ -33,6 +34,7 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
private final SkillVersionRepository skillVersionRepository;
private final ScanTaskProducer scanTaskProducer;
private final ObjectStorageService objectStorageService;
private final int maxRetryAttempts;
public ScanTaskConsumer(RedissonClient redissonClient,
String streamKey,
@ -50,6 +52,7 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
this.skillVersionRepository = skillVersionRepository;
this.scanTaskProducer = scanTaskProducer;
this.objectStorageService = objectStorageService;
this.maxRetryAttempts = 3;
}
public ScanTaskConsumer(RedissonClient redissonClient,
@ -64,6 +67,7 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
Duration reclaimMinIdle,
int reclaimBatchSize,
Duration reclaimInterval,
int maxRetryAttempts,
MessageObservationSupport messageObservationSupport) {
super(
redissonClient,
@ -81,6 +85,30 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
this.skillVersionRepository = skillVersionRepository;
this.scanTaskProducer = scanTaskProducer;
this.objectStorageService = objectStorageService;
this.maxRetryAttempts = maxRetryAttempts;
}
@Override
protected int readBatchSize() {
return 1;
}
@Override
protected int maxRetryCount() {
return maxRetryAttempts;
}
@Override
protected boolean shouldDeferFailure(ScanTaskPayload payload, Exception error) {
return error instanceof ConcurrentScanInProgressException
|| (error instanceof SecurityScanException scanError && scanError.isScannerUnavailable());
}
@Override
protected void markDeferred(ScanTaskPayload payload, Exception error) {
cleanupRetryTempPath(payload);
log.warn("Scanner unavailable; keeping task pending for later recovery: taskId={}, versionId={}, reason={}",
payload.taskId(), payload.versionId(), error.getMessage());
}
@Override

View file

@ -41,7 +41,7 @@ skillhub:
mode: ${SKILLHUB_SECURITY_SCANNER_MODE:upload}
stream:
reclaim-enabled: ${SKILLHUB_SCAN_STREAM_RECLAIM_ENABLED:true}
reclaim-min-idle: ${SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE:PT2M}
reclaim-min-idle: ${SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE:PT16M}
reclaim-batch-size: ${SKILLHUB_SCAN_STREAM_RECLAIM_BATCH_SIZE:20}
reclaim-interval: ${SKILLHUB_SCAN_STREAM_RECLAIM_INTERVAL:PT30S}
bootstrap:

View file

@ -209,7 +209,9 @@ skillhub:
scan-path: /scan-upload
mode: ${SKILLHUB_SECURITY_SCANNER_MODE:local}
connect-timeout-ms: ${SKILLHUB_SECURITY_SCANNER_CONNECT_TIMEOUT:5000}
read-timeout-ms: ${SKILLHUB_SECURITY_SCANNER_READ_TIMEOUT:300000}
# Keep this above the longest expected scan. Availability failures remain pending
# and are reclaimed later instead of moving the skill to SCAN_FAILED.
read-timeout-ms: ${SKILLHUB_SECURITY_SCANNER_READ_TIMEOUT:900000}
retry-max-attempts: ${SKILLHUB_SECURITY_SCANNER_RETRY_MAX:3}
analyzers:
behavioral: ${SKILLHUB_SCANNER_USE_BEHAVIORAL:true}
@ -229,7 +231,8 @@ skillhub:
key: ${SKILLHUB_SCAN_STREAM_KEY:skillhub:scan:requests}
group: ${SKILLHUB_SCAN_STREAM_GROUP:skillhub-scanners}
reclaim-enabled: ${SKILLHUB_SCAN_STREAM_RECLAIM_ENABLED:true}
reclaim-min-idle: ${SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE:PT2M}
# Must exceed scanner read-timeout so an active task is not reclaimed prematurely.
reclaim-min-idle: ${SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE:PT16M}
reclaim-batch-size: ${SKILLHUB_SCAN_STREAM_RECLAIM_BATCH_SIZE:20}
reclaim-interval: ${SKILLHUB_SCAN_STREAM_RECLAIM_INTERVAL:PT30S}
bootstrap:

View file

@ -0,0 +1,25 @@
package com.iflytek.skillhub.config;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import java.util.List;
import org.junit.jupiter.api.Test;
import org.springframework.data.redis.core.StringRedisTemplate;
class SessionRecoveryConfigTest {
@Test
void corruptSessionRemover_deletesSessionKeysWithoutReadingTheCorruptValue() {
StringRedisTemplate redisTemplate = mock(StringRedisTemplate.class);
var remover = new SessionRecoveryConfig()
.corruptSessionRemover(redisTemplate, "skillhub:session");
remover.remove("broken-session");
verify(redisTemplate).delete(List.of(
"skillhub:session:sessions:broken-session",
"skillhub:session:sessions:expires:broken-session"
));
}
}

View file

@ -27,6 +27,7 @@ import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.when;
class AbstractStreamConsumerTest {
@ -35,12 +36,44 @@ class AbstractStreamConsumerTest {
void handleMessage_acknowledgesAfterSuccessfulProcessing() {
@SuppressWarnings("unchecked")
RStream<String, String> stream = mock(RStream.class);
TestConsumer consumer = new TestConsumer(stream);
StreamMessageId messageId = new StreamMessageId(1, 0);
when(stream.ack("scan-group", messageId)).thenReturn(1L);
TestConsumer consumer = new TestConsumer(stream);
consumer.handleMessage(messageId, Map.of("payload", "ok"));
verify(stream).ack("scan-group", messageId);
verify(stream).remove(messageId);
}
@Test
void handleMessage_doesNotDeleteWhenAcknowledgementReturnsZero() {
@SuppressWarnings("unchecked")
RStream<String, String> stream = mock(RStream.class);
StreamMessageId messageId = new StreamMessageId(10, 0);
when(stream.ack("scan-group", messageId)).thenReturn(0L);
TestConsumer consumer = new TestConsumer(stream);
consumer.handleMessage(messageId, Map.of("payload", "ok"));
verify(stream).ack("scan-group", messageId);
verify(stream, never()).remove(messageId);
}
@Test
void handleMessage_doesNotDeleteWhenAcknowledgementFails() {
@SuppressWarnings("unchecked")
RStream<String, String> stream = mock(RStream.class);
StreamMessageId messageId = new StreamMessageId(11, 0);
when(stream.ack("scan-group", messageId))
.thenThrow(new RedisSystemException("redis unavailable", new IllegalStateException("offline")));
TestConsumer consumer = new TestConsumer(stream);
org.assertj.core.api.Assertions.assertThatThrownBy(
() -> consumer.handleMessage(messageId, Map.of("payload", "ok")))
.isInstanceOf(RedisSystemException.class);
verify(stream, never()).remove(messageId);
}
@Test
@ -57,6 +90,22 @@ class AbstractStreamConsumerTest {
verify(stream, times(1)).ack("scan-group", messageId);
}
@Test
void handleMessage_deferredFailureRemainsPending() {
@SuppressWarnings("unchecked")
RStream<String, String> stream = mock(RStream.class);
TestConsumer consumer = new TestConsumer(stream);
consumer.fail = true;
consumer.defer = true;
StreamMessageId messageId = new StreamMessageId(20, 0);
consumer.handleMessage(messageId, Map.of("payload", "busy"));
verify(stream, never()).ack("scan-group", messageId);
verify(stream, never()).remove(messageId);
assertThat(consumer.deferred).isTrue();
}
@Test
void consumeAvailableMessages_processesNeverDeliveredMessages() {
@SuppressWarnings("unchecked")
@ -140,6 +189,8 @@ class AbstractStreamConsumerTest {
private final RStream<String, String> stream;
private final RequestIdAccessor requestIdAccessor;
private boolean fail;
private boolean defer;
private boolean deferred;
private String processedRequestId;
private TestConsumer(RStream<String, String> stream) {
@ -209,6 +260,16 @@ class AbstractStreamConsumerTest {
@Override
protected void retryMessage(String payload, int retryCount) {
}
@Override
protected boolean shouldDeferFailure(String payload, Exception error) {
return defer;
}
@Override
protected void markDeferred(String payload, Exception error) {
deferred = true;
}
}
private static final class CountingConsumer extends TestConsumer {

View file

@ -18,6 +18,8 @@ import com.iflytek.skillhub.observability.MessageObservationSupport;
import com.iflytek.skillhub.observability.RequestIdAccessor;
import com.iflytek.skillhub.storage.ObjectStorageService;
import com.iflytek.skillhub.storage.ObjectMetadata;
import com.iflytek.skillhub.infra.http.HttpClientException;
import com.iflytek.skillhub.infra.scanner.SecurityScanException;
import io.micrometer.observation.ObservationRegistry;
import org.junit.jupiter.api.Test;
import org.redisson.api.RLock;
@ -307,7 +309,7 @@ class ScanTaskConsumerTest {
}
@Test
void handleMessage_whenTaskLockIsHeld_republishesInsteadOfDroppingDelivery() {
void handleMessage_whenTaskLockIsHeld_keepsOriginalDeliveryPending() {
StubSecurityScanner securityScanner = new StubSecurityScanner();
InMemoryScanTaskProducer producer = new InMemoryScanTaskProducer();
RLock processingLock = mock(RLock.class);
@ -328,9 +330,44 @@ class ScanTaskConsumerTest {
"scannerType", ScannerType.SKILL_SCANNER.getValue()
));
assertThat(producer.publishedTask.taskId()).isEqualTo("task-reclaimed");
assertThat(producer.publishedTask.metadata()).containsEntry("retryCount", "1");
verify(consumer.stream).ack("skillhub-scanners", new StreamMessageId(11, 0));
assertThat(producer.publishedTask).isNull();
verify(consumer.stream, never()).ack("skillhub-scanners", new StreamMessageId(11, 0));
}
@Test
void handleMessage_whenScannerIsUnavailable_keepsVersionScanningAndDeliveryPending() {
StubSecurityScanner securityScanner = new StubSecurityScanner();
securityScanner.failure = new SecurityScanException(
"scanner timed out", new HttpClientException("request timed out", new java.util.concurrent.TimeoutException()));
SkillVersion version = new SkillVersion(8L, "1.0.0", "publisher-1");
try {
setField(version, "id", 42L);
} catch (Exception e) {
throw new AssertionError(e);
}
version.setStatus(SkillVersionStatus.SCANNING);
InMemorySkillVersionRepository repository = new InMemorySkillVersionRepository(version);
InMemoryScanTaskProducer producer = new InMemoryScanTaskProducer();
TestableScanTaskConsumer consumer = new TestableScanTaskConsumer(
securityScanner,
new StubSecurityScanService(),
repository,
producer,
new InMemoryObjectStorageService()
);
StreamMessageId messageId = new StreamMessageId(12, 0);
consumer.handleMessage(messageId, Map.of(
"taskId", "task-timeout",
"versionId", "42",
"skillPath", "/tmp/skillhub-scans/42",
"scannerType", ScannerType.SKILL_SCANNER.getValue()
));
assertThat(version.getStatus()).isEqualTo(SkillVersionStatus.SCANNING);
assertThat(repository.savedVersion).isNull();
assertThat(producer.publishedTask).isNull();
verify(consumer.stream, never()).ack("skillhub-scanners", messageId);
}
@Test

View file

@ -8,6 +8,7 @@ import com.iflytek.skillhub.auth.oauth.SkillHubOAuth2AuthorizationRequestResolve
import com.iflytek.skillhub.auth.mock.MockAuthFilter;
import com.iflytek.skillhub.auth.policy.RouteSecurityPolicyRegistry;
import com.iflytek.skillhub.auth.session.ExpiredPublicSessionFilter;
import com.iflytek.skillhub.auth.session.CorruptSessionRemover;
import com.iflytek.skillhub.auth.token.ApiTokenAuthenticationFilter;
import com.iflytek.skillhub.auth.token.ApiTokenScopeFilter;
import jakarta.servlet.http.Cookie;
@ -16,6 +17,7 @@ import jakarta.servlet.http.HttpServletResponse;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
@ -68,6 +70,8 @@ public class SecurityConfig {
private final AccessDeniedHandler apiAccessDeniedHandler;
private final ObjectProvider<MockAuthFilter> mockAuthFilterProvider;
private final RouteSecurityPolicyRegistry routeSecurityPolicyRegistry;
private final CorruptSessionRemover corruptSessionRemover;
private final String sessionCookieName;
public SecurityConfig(CustomOAuth2UserService customOAuth2UserService,
CustomOidcUserService customOidcUserService,
@ -79,7 +83,9 @@ public class SecurityConfig {
AuthenticationEntryPoint apiAuthenticationEntryPoint,
AccessDeniedHandler apiAccessDeniedHandler,
ObjectProvider<MockAuthFilter> mockAuthFilterProvider,
RouteSecurityPolicyRegistry routeSecurityPolicyRegistry) {
RouteSecurityPolicyRegistry routeSecurityPolicyRegistry,
ObjectProvider<CorruptSessionRemover> corruptSessionRemoverProvider,
@Value("${server.servlet.session.cookie.name:SESSION}") String sessionCookieName) {
this.customOAuth2UserService = customOAuth2UserService;
this.customOidcUserService = customOidcUserService;
this.authorizationRequestResolver = authorizationRequestResolver;
@ -91,6 +97,10 @@ public class SecurityConfig {
this.apiAccessDeniedHandler = apiAccessDeniedHandler;
this.mockAuthFilterProvider = mockAuthFilterProvider;
this.routeSecurityPolicyRegistry = routeSecurityPolicyRegistry;
this.corruptSessionRemover = corruptSessionRemoverProvider.getIfAvailable(() -> sessionId -> {
throw new IllegalStateException("Corrupt session recovery is not configured");
});
this.sessionCookieName = sessionCookieName;
}
/**
@ -161,9 +171,10 @@ public class SecurityConfig {
response.sendRedirect(((contextPath == null) ? "" : contextPath) + "/");
})
.invalidateHttpSession(true)
.deleteCookies("SESSION")
.deleteCookies(sessionCookieName)
)
.addFilterBefore(new ExpiredPublicSessionFilter(routeSecurityPolicyRegistry), CsrfFilter.class)
.addFilterBefore(new ExpiredPublicSessionFilter(
routeSecurityPolicyRegistry, corruptSessionRemover, sessionCookieName), CsrfFilter.class)
.addFilterBefore(apiTokenAuthenticationFilter, UsernamePasswordAuthenticationFilter.class)
.addFilterAfter(apiTokenScopeFilter, ApiTokenAuthenticationFilter.class);

View file

@ -0,0 +1,11 @@
package com.iflytek.skillhub.auth.session;
/**
* Removes a session record that cannot be deserialized by the current application version.
* The input is the server-side session id already resolved by Spring Session.
*/
@FunctionalInterface
public interface CorruptSessionRemover {
void remove(String sessionId);
}

View file

@ -9,7 +9,11 @@ import jakarta.servlet.http.HttpServletRequestWrapper;
import jakarta.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.web.filter.OncePerRequestFilter;
/**
@ -17,12 +21,21 @@ import org.springframework.web.filter.OncePerRequestFilter;
*/
public final class ExpiredPublicSessionFilter extends OncePerRequestFilter {
private static final Set<String> SESSION_COOKIES = Set.of("SESSION", "JSESSIONID");
private static final Logger log = LoggerFactory.getLogger(ExpiredPublicSessionFilter.class);
private final RouteSecurityPolicyRegistry routeSecurityPolicyRegistry;
private final CorruptSessionRemover corruptSessionRemover;
private final Set<String> sessionCookieNames;
public ExpiredPublicSessionFilter(RouteSecurityPolicyRegistry routeSecurityPolicyRegistry) {
public ExpiredPublicSessionFilter(RouteSecurityPolicyRegistry routeSecurityPolicyRegistry,
CorruptSessionRemover corruptSessionRemover,
String sessionCookieName) {
this.routeSecurityPolicyRegistry = routeSecurityPolicyRegistry;
this.corruptSessionRemover = corruptSessionRemover;
Set<String> cookieNames = new HashSet<>();
cookieNames.add(sessionCookieName);
cookieNames.add("JSESSIONID");
this.sessionCookieNames = Set.copyOf(cookieNames);
}
@Override
@ -33,17 +46,55 @@ public final class ExpiredPublicSessionFilter extends OncePerRequestFilter {
String requestPath = RouteSecurityPolicyRegistry.requestPath(request);
boolean publicRoute = routeSecurityPolicyRegistry.accessLevel(request.getMethod(), requestPath)
== RouteSecurityPolicyRegistry.AccessLevel.PERMIT_ALL;
if (publicRoute && request.getRequestedSessionId() != null && !request.isRequestedSessionIdValid()) {
filterChain.doFilter(new SessionlessRequest(request), response);
try {
String requestedSessionId = request.getRequestedSessionId();
if (publicRoute && requestedSessionId != null && !request.isRequestedSessionIdValid()) {
filterChain.doFilter(new SessionlessRequest(request, sessionCookieNames), response);
return;
}
} catch (SerializationException error) {
// Spring Session resolves and caches the server-side session id before loading the
// Redis record. Re-read that cached id so custom cookie serializers and jvmRoute
// settings remain Spring Session's responsibility.
String corruptSessionId = request.getRequestedSessionId();
if (corruptSessionId == null) {
throw error;
}
// Delete only the unreadable record. Redis connectivity failures from deleteById
// intentionally propagate instead of disguising a storage outage as logout.
corruptSessionRemover.remove(corruptSessionId);
expireSessionCookies(request, response);
log.warn("Removed an unreadable HTTP session; the client must authenticate again");
filterChain.doFilter(new SessionlessRequest(request, sessionCookieNames), response);
return;
}
filterChain.doFilter(request, response);
}
private void expireSessionCookies(HttpServletRequest request, HttpServletResponse response) {
Cookie[] cookies = request.getCookies();
if (cookies == null) {
return;
}
Arrays.stream(cookies)
.filter(cookie -> sessionCookieNames.contains(cookie.getName()))
.forEach(cookie -> {
Cookie expired = new Cookie(cookie.getName(), "");
expired.setHttpOnly(true);
expired.setSecure(request.isSecure());
expired.setPath(request.getContextPath().isBlank() ? "/" : request.getContextPath());
expired.setMaxAge(0);
response.addCookie(expired);
});
}
private static final class SessionlessRequest extends HttpServletRequestWrapper {
private SessionlessRequest(HttpServletRequest request) {
private final Set<String> sessionCookieNames;
private SessionlessRequest(HttpServletRequest request, Set<String> sessionCookieNames) {
super(request);
this.sessionCookieNames = sessionCookieNames;
}
@Override
@ -68,7 +119,7 @@ public final class ExpiredPublicSessionFilter extends OncePerRequestFilter {
return null;
}
Cookie[] retained = Arrays.stream(cookies)
.filter(cookie -> !SESSION_COOKIES.contains(cookie.getName()))
.filter(cookie -> !sessionCookieNames.contains(cookie.getName()))
.toArray(Cookie[]::new);
return retained.length == 0 ? null : retained;
}

View file

@ -2,10 +2,13 @@ package com.iflytek.skillhub.auth.session;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import com.iflytek.skillhub.auth.policy.RouteSecurityPolicyRegistry;
import jakarta.servlet.FilterChain;
@ -13,15 +16,19 @@ import jakarta.servlet.ServletRequest;
import jakarta.servlet.ServletResponse;
import jakarta.servlet.http.Cookie;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletRequestWrapper;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.mock.web.MockHttpServletRequest;
import org.springframework.mock.web.MockHttpServletResponse;
import org.springframework.data.redis.serializer.SerializationException;
class ExpiredPublicSessionFilterTest {
private final ExpiredPublicSessionFilter filter =
new ExpiredPublicSessionFilter(new RouteSecurityPolicyRegistry());
private final CorruptSessionRemover corruptSessionRemover = mock(CorruptSessionRemover.class);
private final ExpiredPublicSessionFilter filter = new ExpiredPublicSessionFilter(
new RouteSecurityPolicyRegistry(), corruptSessionRemover, "SESSION");
@Test
void expiredSessionOnPublicRoute_shouldBeHiddenFromDownstreamSecurityFilters() throws Exception {
@ -74,6 +81,61 @@ class ExpiredPublicSessionFilterTest {
assertNull(capturedRequest(chain).getRequestedSessionId());
}
@Test
void unreadableSession_shouldBeDeletedAndTreatedAsLoggedOut() throws Exception {
MockHttpServletRequest delegate = new MockHttpServletRequest();
delegate.setMethod("GET");
delegate.setRequestURI("/api/v1/auth/methods");
delegate.setCookies(new Cookie("SESSION", "corrupt-session"));
AtomicInteger calls = new AtomicInteger();
HttpServletRequest request = new HttpServletRequestWrapper(delegate) {
@Override
public String getRequestedSessionId() {
if (calls.getAndIncrement() == 0) {
throw new SerializationException("incompatible session data");
}
return "resolved-session-id";
}
};
MockHttpServletResponse response = new MockHttpServletResponse();
FilterChain chain = mock(FilterChain.class);
filter.doFilter(request, response, chain);
verify(corruptSessionRemover).remove("resolved-session-id");
assertNull(capturedRequest(chain).getRequestedSessionId());
assertNotNull(response.getCookie("SESSION"));
assertEquals(0, response.getCookie("SESSION").getMaxAge());
}
@Test
void unreadableSession_shouldNotHideRedisDeleteFailure() {
MockHttpServletRequest delegate = new MockHttpServletRequest();
delegate.setMethod("GET");
delegate.setRequestURI("/api/v1/auth/methods");
delegate.setCookies(new Cookie("SESSION", "encoded-cookie"));
AtomicInteger calls = new AtomicInteger();
HttpServletRequest request = new HttpServletRequestWrapper(delegate) {
@Override
public String getRequestedSessionId() {
if (calls.getAndIncrement() == 0) {
throw new SerializationException("incompatible session data");
}
return "resolved-session-id";
}
};
RuntimeException redisFailure = new RuntimeException("redis unavailable");
org.mockito.Mockito.doThrow(redisFailure)
.when(corruptSessionRemover).remove("resolved-session-id");
FilterChain chain = mock(FilterChain.class);
RuntimeException actual = assertThrows(RuntimeException.class,
() -> filter.doFilter(request, new MockHttpServletResponse(), chain));
assertSame(redisFailure, actual);
verifyNoInteractions(chain);
}
private static MockHttpServletRequest expiredSessionRequest(String method, String path) {
MockHttpServletRequest request = new MockHttpServletRequest();
request.setMethod(method);

View file

@ -12,7 +12,7 @@ public class HttpClientException extends RuntimeException {
}
public HttpClientException(String message, Throwable cause) {
super(message, cause);
super(message + ": " + rootCauseSummary(cause), cause);
this.statusCode = 0;
this.responseBody = null;
}
@ -24,4 +24,13 @@ public class HttpClientException extends RuntimeException {
public String getResponseBody() {
return responseBody;
}
private static String rootCauseSummary(Throwable error) {
Throwable root = error;
while (root.getCause() != null && root.getCause() != root) {
root = root.getCause();
}
String message = root.getMessage();
return root.getClass().getSimpleName() + (message == null || message.isBlank() ? "" : ": " + message);
}
}

View file

@ -9,4 +9,15 @@ public class SecurityScanException extends RuntimeException {
public SecurityScanException(String message) {
super(message);
}
/**
* Returns true when retrying later is safer than permanently failing the skill version.
*/
public boolean isScannerUnavailable() {
if (!(getCause() instanceof com.iflytek.skillhub.infra.http.HttpClientException error)) {
return false;
}
int status = error.getStatusCode();
return status == 0 || status == 429 || status >= 500;
}
}

View file

@ -40,7 +40,7 @@ public class SkillScannerAdapter implements SecurityScanner {
return mapToResponse(apiResponse);
} catch (HttpClientException e) {
log.error("Security scan failed for versionId={}: {}", request.skillVersionId(), e.getMessage());
throw new SecurityScanException("Security scan failed", e);
throw new SecurityScanException("Security scan request failed: " + e.getMessage(), e);
}
}

View file

@ -7,6 +7,8 @@ import com.iflytek.skillhub.domain.security.SecurityVerdict;
import com.iflytek.skillhub.infra.http.HttpClient;
import com.iflytek.skillhub.infra.http.HttpClientException;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.springframework.http.HttpHeaders;
import java.nio.file.Path;
@ -87,16 +89,45 @@ class SkillScannerAdapterTest {
assertThat(response.verdict()).isEqualTo(SecurityVerdict.BLOCKED);
}
@Test
void scan_wrapsHttpClientFailureAsSecurityScanException() {
@ParameterizedTest
@ValueSource(ints = {429, 500, 599})
void scan_treatsTransientHttpFailureAsScannerUnavailable(int statusCode) {
StubSkillScannerService skillScannerService = new StubSkillScannerService();
skillScannerService.directoryException = new HttpClientException(502, "bad gateway");
skillScannerService.directoryException = new HttpClientException(statusCode, "scanner unavailable");
ScanOptions options = ScanOptions.disabled();
SkillScannerAdapter adapter = new SkillScannerAdapter(skillScannerService, "local", options);
assertThatThrownBy(() -> adapter.scan(new SecurityScanRequest("task-1", 42L, "/tmp/skill", Map.of())))
.isInstanceOf(SecurityScanException.class)
.hasMessage("Security scan failed");
.hasMessage("Security scan request failed: HTTP " + statusCode + ": scanner unavailable")
.satisfies(error -> assertThat(((SecurityScanException) error).isScannerUnavailable()).isTrue());
}
@Test
void scan_treatsConnectionFailureAsScannerUnavailable() {
StubSkillScannerService skillScannerService = new StubSkillScannerService();
skillScannerService.directoryException =
new HttpClientException("request failed", new IllegalStateException("connection refused"));
SkillScannerAdapter adapter = new SkillScannerAdapter(
skillScannerService, "local", ScanOptions.disabled());
assertThatThrownBy(() -> adapter.scan(new SecurityScanRequest("task-1", 42L, "/tmp/skill", Map.of())))
.isInstanceOf(SecurityScanException.class)
.satisfies(error -> assertThat(((SecurityScanException) error).isScannerUnavailable()).isTrue());
}
@ParameterizedTest
@ValueSource(ints = {400, 422, 499})
void scan_treatsDeterministicClientFailureAsPermanent(int statusCode) {
StubSkillScannerService skillScannerService = new StubSkillScannerService();
skillScannerService.directoryException = new HttpClientException(statusCode, "invalid package");
SkillScannerAdapter adapter = new SkillScannerAdapter(
skillScannerService, "local", ScanOptions.disabled());
assertThatThrownBy(() -> adapter.scan(
new SecurityScanRequest("task-1", 42L, "/tmp/skill", Map.of())))
.isInstanceOf(SecurityScanException.class)
.satisfies(error -> assertThat(((SecurityScanException) error).isScannerUnavailable()).isFalse());
}
private static final class StubSkillScannerService extends SkillScannerService {

View file

@ -111,12 +111,20 @@ async function cacheAccountSession(page: Page, username: string) {
})
}
async function restoreCachedSession(page: Page, worker: number): Promise<SessionSnapshot | null> {
async function restoreCachedSession(
page: Page,
worker: number,
allowMockSession = true,
): Promise<SessionSnapshot | null> {
const snapshot = cachedSessionByWorker.get(worker)
if (!snapshot) {
return null
}
if (!allowMockSession && snapshot.username === 'local-user') {
return null
}
await page.context().addCookies(snapshot.cookies)
if (await hasActiveSession(page)) {
return snapshot
@ -170,20 +178,22 @@ async function tryBootstrapMockSession(page: Page, worker: number): Promise<{ us
async function registerSessionOnce(page: Page, testInfo?: TestInfo, options?: RegisterSessionOptions) {
const worker = testInfo?.parallelIndex ?? 0
const cached = cachedUserByWorker.get(worker)
const allowMockSession = options?.allowMockSession !== false
const cachedUsername = cachedUserByWorker.get(worker)
const cached = !allowMockSession && cachedUsername === 'local-user' ? undefined : cachedUsername
const username = usernameForWorker(testInfo)
const request = page.context().request
await primeAuthProviders(page)
// Avoid hammering auth endpoints on every test run for the same worker.
const restored = await restoreCachedSession(page, worker)
const restored = await restoreCachedSession(page, worker, allowMockSession)
if (restored) {
cachedUserByWorker.set(worker, restored.username)
return { username: restored.username, password }
}
if (options?.allowMockSession !== false) {
if (allowMockSession) {
const mockSession = await tryBootstrapMockSession(page, worker)
if (mockSession) {
return mockSession

View file

@ -44,7 +44,9 @@ test.describe('Public Skill Detail Anonymous Access (Real API)', () => {
await expect(page).toHaveURL(new RegExp(`/space/${current.skill.namespace}/${current.skill.slug}(\\?|$)`))
await expect(page).not.toHaveURL(/\/login\?returnTo=/)
await expect(page.getByRole('heading', { name: current.skillName, exact: true })).toBeVisible()
const skillNameHeadings = page.getByRole('heading', { name: current.skillName, exact: true })
await expect(skillNameHeadings).toHaveCount(2)
await expect(skillNameHeadings.first()).toBeVisible()
await expect(page.getByText('Install', { exact: true })).toBeVisible()
const clawhubTarget = current.skill.namespace === 'global'
? current.skill.slug
@ -53,14 +55,13 @@ test.describe('Public Skill Detail Anonymous Access (Real API)', () => {
? ''
: ` --namespace ${current.skill.namespace}`
await expect(page.getByRole('tab', { name: 'ClawHub CLI' })).toHaveAttribute('aria-selected', 'true')
await expect(page.getByText(new RegExp(`npx clawhub install ${escapeRegExp(clawhubTarget)} --registry`))).toBeVisible()
await expect(page.getByRole('tab', { name: 'SkillHub CLI' })).toBeVisible()
await page.getByRole('tab', { name: 'SkillHub CLI' }).click()
await expect(page.getByRole('tab', { name: 'SkillHub CLI' })).toHaveAttribute('aria-selected', 'true')
await expect(page.getByText(new RegExp(`npx @astron-team/skillhub@latest install ${escapeRegExp(current.skill.slug)}${escapeRegExp(skillhubNamespace)} --registry`))).toBeVisible()
await expect(page.getByRole('button', { name: 'Copy' }).first()).toBeVisible()
await page.getByRole('tab', { name: 'ClawHub CLI' }).click()
await expect(page.getByRole('tab', { name: 'ClawHub CLI' })).toHaveAttribute('aria-selected', 'true')
await expect(page.getByText(new RegExp(`npx clawhub install ${escapeRegExp(clawhubTarget)} --registry`))).toBeVisible()
})
})

View file

@ -1,13 +1,13 @@
import { expect, test } from '@playwright/test'
import { setEnglishLocale } from './helpers/auth-fixtures'
import { createFreshSession } from './helpers/session'
import { registerSession } from './helpers/session'
test.describe('Settings Pages (Real API)', () => {
test.use({ baseURL: 'http://127.0.0.1:3000' })
test.beforeEach(async ({ page }, testInfo) => {
await setEnglishLocale(page)
await createFreshSession(page, testInfo)
await registerSession(page, testInfo, { allowMockSession: false })
})
test('opens profile settings page', async ({ page }) => {

View file

@ -125,7 +125,7 @@ describe('install-command', () => {
expect(html).not.toContain('flex-1 rounded-md')
})
it('renders ClawHub CLI as the default install method', () => {
it('renders SkillHub CLI as the default install method', () => {
setMockWindow('https://app.example.com')
const html = renderToStaticMarkup(createElement(InstallCommand, {
@ -136,7 +136,7 @@ describe('install-command', () => {
expect(html).toContain('skillDetail.installMethodClawhub')
expect(html).toContain('skillDetail.installMethodSkillhub')
expect(html).toContain('aria-selected="true"')
expect(html).toContain('npx clawhub install team-alpha--meeting-minutes-generator --registry https://app.example.com')
expect(html).not.toContain('npx @astron-team/skillhub@latest install meeting-minutes-generator --namespace team-alpha --registry https://app.example.com')
expect(html).toContain('npx @astron-team/skillhub@latest install meeting-minutes-generator --namespace team-alpha --registry https://app.example.com')
expect(html).not.toContain('npx clawhub install team-alpha--meeting-minutes-generator --registry https://app.example.com')
})
})

View file

@ -85,21 +85,21 @@ export function InstallCommand({ namespace, slug }: InstallCommandProps) {
const skillhubCommand = useMemo(() => buildSkillhubInstallCommand(namespace, slug, baseUrl), [baseUrl, namespace, slug])
return (
<Tabs defaultValue="clawhub" className="space-y-3">
<Tabs defaultValue="skillhub" className="space-y-3">
<TabsList className="w-full gap-6 border-border/70 bg-transparent p-0 text-xs">
<TabsTrigger value="clawhub" className={installMethodTabTriggerClass}>
{t('skillDetail.installMethodClawhub')}
</TabsTrigger>
<TabsTrigger value="skillhub" className={installMethodTabTriggerClass}>
{t('skillDetail.installMethodSkillhub')}
</TabsTrigger>
<TabsTrigger value="clawhub" className={installMethodTabTriggerClass}>
{t('skillDetail.installMethodClawhub')}
</TabsTrigger>
</TabsList>
<TabsContent value="clawhub">
<CommandBlock command={clawhubCommand} />
</TabsContent>
<TabsContent value="skillhub">
<CommandBlock command={skillhubCommand} />
</TabsContent>
<TabsContent value="clawhub">
<CommandBlock command={clawhubCommand} />
</TabsContent>
</Tabs>
)
}