fix(ssf): 修复安全事件连接基础阻断
为已记录旧迁移版本的环境补充幂等表修复迁移,并让 Transmitter 网络连接在 IPv6 地址不可达时继续尝试 IPv4。失败连接现在通过 If-Match 安全重试,页面同步提供重试入口并清理过期错误提示。\n\n验证:Go securityevents/migrate 测试、Web Vitest 与 TypeScript 类型检查通过。
This commit is contained in:
@@ -0,0 +1,23 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"os"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestSecurityEventIdempotencyRepairMigrationExists(t *testing.T) {
|
||||||
|
payload, err := os.ReadFile("../../migrations/0064_security_event_connection_idempotency_repair.sql")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("read repair migration: %v", err)
|
||||||
|
}
|
||||||
|
sql := string(payload)
|
||||||
|
for _, statement := range []string{
|
||||||
|
"CREATE TABLE IF NOT EXISTS gateway_security_event_connection_idempotency",
|
||||||
|
"CREATE INDEX IF NOT EXISTS idx_gateway_security_event_connection_idempotency_created",
|
||||||
|
} {
|
||||||
|
if !strings.Contains(sql, statement) {
|
||||||
|
t.Fatalf("repair migration is missing %q", statement)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -879,11 +879,26 @@ func safeHTTPClient(issuer *url.URL, appEnv string) *http.Client {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
dialer := &net.Dialer{Timeout: 5 * time.Second}
|
dialer := &net.Dialer{Timeout: 5 * time.Second}
|
||||||
return dialer.DialContext(ctx, network, net.JoinHostPort(addresses[0].IP.String(), port))
|
return dialResolvedAddresses(ctx, network, port, addresses, dialer)
|
||||||
}
|
}
|
||||||
return &http.Client{Timeout: 10 * time.Second, Transport: transport, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
|
return &http.Client{Timeout: 10 * time.Second, Transport: transport, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func dialResolvedAddresses(ctx context.Context, network, port string, addresses []net.IPAddr, dialer *net.Dialer) (net.Conn, error) {
|
||||||
|
var attempts []error
|
||||||
|
for _, address := range addresses {
|
||||||
|
connection, err := dialer.DialContext(ctx, network, net.JoinHostPort(address.IP.String(), port))
|
||||||
|
if err == nil {
|
||||||
|
return connection, nil
|
||||||
|
}
|
||||||
|
attempts = append(attempts, err)
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil, fmt.Errorf("transmitter connection failed for every resolved address: %w", errors.Join(attempts...))
|
||||||
|
}
|
||||||
|
|
||||||
func blockedAddress(ip net.IP) bool {
|
func blockedAddress(ip net.IP) bool {
|
||||||
if ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() ||
|
if ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() ||
|
||||||
ip.IsUnspecified() || ip.IsMulticast() {
|
ip.IsUnspecified() || ip.IsMulticast() {
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
@@ -148,6 +149,39 @@ func TestTransmitterSSRFProtectionBlocksReservedNetworks(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestDialResolvedAddressesFallsBackFromIPv6ToIPv4(t *testing.T) {
|
||||||
|
listener, err := net.Listen("tcp4", "127.0.0.1:0")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer listener.Close()
|
||||||
|
port := listener.Addr().(*net.TCPAddr).Port
|
||||||
|
accepted := make(chan struct{})
|
||||||
|
go func() {
|
||||||
|
connection, acceptErr := listener.Accept()
|
||||||
|
if acceptErr == nil {
|
||||||
|
_ = connection.Close()
|
||||||
|
close(accepted)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
connection, err := dialResolvedAddresses(ctx, "tcp", strconv.Itoa(port), []net.IPAddr{
|
||||||
|
{IP: net.ParseIP("::1")},
|
||||||
|
{IP: net.ParseIP("127.0.0.1")},
|
||||||
|
}, &net.Dialer{Timeout: 250 * time.Millisecond})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("IPv4 fallback was not attempted: %v", err)
|
||||||
|
}
|
||||||
|
_ = connection.Close()
|
||||||
|
select {
|
||||||
|
case <-accepted:
|
||||||
|
case <-ctx.Done():
|
||||||
|
t.Fatal("IPv4 listener did not receive the fallback connection")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestRetiringConnectionStillAppliesRevocationWatermark(t *testing.T) {
|
func TestRetiringConnectionStillAppliesRevocationWatermark(t *testing.T) {
|
||||||
audience := "urn:easyai:ssf:receiver:" + uuid.NewString()
|
audience := "urn:easyai:ssf:receiver:" + uuid.NewString()
|
||||||
repository := &memoryConnectionRepository{
|
repository := &memoryConnectionRepository{
|
||||||
|
|||||||
@@ -0,0 +1,15 @@
|
|||||||
|
-- 0063 was released before its idempotency table was added. Installations that
|
||||||
|
-- already recorded 0063 need a new immutable migration to repair that drift.
|
||||||
|
CREATE TABLE IF NOT EXISTS gateway_security_event_connection_idempotency (
|
||||||
|
operation text NOT NULL,
|
||||||
|
idempotency_key text NOT NULL,
|
||||||
|
request_hash text NOT NULL,
|
||||||
|
response jsonb NOT NULL,
|
||||||
|
created_at timestamptz NOT NULL DEFAULT now(),
|
||||||
|
PRIMARY KEY (operation, idempotency_key),
|
||||||
|
CONSTRAINT gateway_security_event_connection_idempotency_operation
|
||||||
|
CHECK (operation IN ('connect','verify','rotate','disconnect'))
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_gateway_security_event_connection_idempotency_created
|
||||||
|
ON gateway_security_event_connection_idempotency(created_at);
|
||||||
+13
-3
@@ -1033,7 +1033,8 @@ export function App() {
|
|||||||
setCoreState('loading');
|
setCoreState('loading');
|
||||||
setCoreMessage('');
|
setCoreMessage('');
|
||||||
try {
|
try {
|
||||||
const connection = await connectSecurityEventTransmitter(token, transmitterIssuer);
|
const version = securityEventConnection?.connection?.version;
|
||||||
|
const connection = await connectSecurityEventTransmitter(token, transmitterIssuer, version);
|
||||||
setSecurityEventConnection(connection);
|
setSecurityEventConnection(connection);
|
||||||
setCoreState('ready');
|
setCoreState('ready');
|
||||||
setCoreMessage('认证中心安全事件连接正在验证。');
|
setCoreMessage('认证中心安全事件连接正在验证。');
|
||||||
@@ -1045,8 +1046,17 @@ export function App() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async function refreshSecurityEvents() {
|
async function refreshSecurityEvents() {
|
||||||
const connection = await getSecurityEventConnection(token);
|
setCoreState('loading');
|
||||||
setSecurityEventConnection(connection);
|
setCoreMessage('');
|
||||||
|
try {
|
||||||
|
const connection = await getSecurityEventConnection(token);
|
||||||
|
setSecurityEventConnection(connection);
|
||||||
|
setCoreState('ready');
|
||||||
|
} catch (err) {
|
||||||
|
setCoreState('error');
|
||||||
|
setCoreMessage(err instanceof Error ? err.message : '刷新安全事件连接失败');
|
||||||
|
throw err;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async function verifySecurityEvents() {
|
async function verifySecurityEvents() {
|
||||||
|
|||||||
@@ -55,6 +55,21 @@ describe('security event connection transport', () => {
|
|||||||
expect(JSON.parse(String(init.body))).toEqual({ transmitter_issuer: 'https://auth.example/ssf' });
|
expect(JSON.parse(String(init.body))).toEqual({ transmitter_issuer: 'https://auth.example/ssf' });
|
||||||
expect(String(init.body)).not.toMatch(/secret|bearer|scope|credential/i);
|
expect(String(init.body)).not.toMatch(/secret|bearer|scope|credential/i);
|
||||||
expect(new Headers(init.headers).get('Idempotency-Key')).toContain('ssf-connect-');
|
expect(new Headers(init.headers).get('Idempotency-Key')).toContain('ssf-connect-');
|
||||||
|
expect(new Headers(init.headers).has('If-Match')).toBe(false);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('sends If-Match when retrying an existing failed connection', async () => {
|
||||||
|
const fetchMock = vi.fn().mockResolvedValue(new Response(JSON.stringify({ connected: true }), {
|
||||||
|
status: 202,
|
||||||
|
headers: { 'Content-Type': 'application/json' },
|
||||||
|
}));
|
||||||
|
vi.stubGlobal('fetch', fetchMock);
|
||||||
|
vi.stubGlobal('crypto', { randomUUID: () => '00000000-0000-4000-8000-000000000000' });
|
||||||
|
|
||||||
|
await connectSecurityEventTransmitter('manager-token', 'https://auth.example/ssf', 7);
|
||||||
|
|
||||||
|
const [, init] = fetchMock.mock.calls[0] as [string, RequestInit];
|
||||||
|
expect(new Headers(init.headers).get('If-Match')).toBe('W/"7"');
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
+4
-2
@@ -1007,10 +1007,12 @@ export async function getSecurityEventConnection(token: string): Promise<Securit
|
|||||||
return request<SecurityEventConnectionResponse>(securityEventConnectionPath, { token });
|
return request<SecurityEventConnectionResponse>(securityEventConnectionPath, { token });
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function connectSecurityEventTransmitter(token: string, transmitterIssuer: string): Promise<SecurityEventConnectionResponse> {
|
export async function connectSecurityEventTransmitter(token: string, transmitterIssuer: string, version?: number): Promise<SecurityEventConnectionResponse> {
|
||||||
|
const headers: Record<string, string> = { 'Idempotency-Key': `ssf-connect-${crypto.randomUUID()}` };
|
||||||
|
if (version !== undefined) headers['If-Match'] = `W/"${version}"`;
|
||||||
return request<SecurityEventConnectionResponse>(securityEventConnectionPath, {
|
return request<SecurityEventConnectionResponse>(securityEventConnectionPath, {
|
||||||
body: { transmitter_issuer: transmitterIssuer }, method: 'PUT', token,
|
body: { transmitter_issuer: transmitterIssuer }, method: 'PUT', token,
|
||||||
headers: { 'Idempotency-Key': `ssf-connect-${crypto.randomUUID()}` },
|
headers,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -474,6 +474,9 @@ function SecurityEventConnectionPanel(props: {
|
|||||||
{connection.lastErrorCategory && <span>最近错误: {connection.lastErrorCategory}</span>}
|
{connection.lastErrorCategory && <span>最近错误: {connection.lastErrorCategory}</span>}
|
||||||
</div>
|
</div>
|
||||||
<div className="fileStorageToolbar">
|
<div className="fileStorageToolbar">
|
||||||
|
{connection.lifecycleStatus === 'error' && (
|
||||||
|
<Button type="button" size="sm" disabled={props.loading} onClick={() => void props.onConnect(connection.transmitterIssuer)}><Link2 size={14} />重试连接</Button>
|
||||||
|
)}
|
||||||
<Button type="button" variant="outline" size="sm" disabled={props.loading} onClick={() => void props.onVerify()}><RefreshCw size={14} />重新验证</Button>
|
<Button type="button" variant="outline" size="sm" disabled={props.loading} onClick={() => void props.onVerify()}><RefreshCw size={14} />重新验证</Button>
|
||||||
<Button type="button" variant="outline" size="sm" disabled={props.loading || connection.lifecycleStatus === 'rotating'} onClick={() => void props.onRotate()}><RotateCcw size={14} />轮换接收凭据</Button>
|
<Button type="button" variant="outline" size="sm" disabled={props.loading || connection.lifecycleStatus === 'rotating'} onClick={() => void props.onRotate()}><RotateCcw size={14} />轮换接收凭据</Button>
|
||||||
<Button type="button" variant="destructive" size="sm" disabled={props.loading || connection.lifecycleStatus === 'retiring'} onClick={props.onDisconnect}><Unplug size={14} />断开连接</Button>
|
<Button type="button" variant="destructive" size="sm" disabled={props.loading || connection.lifecycleStatus === 'retiring'} onClick={props.onDisconnect}><Unplug size={14} />断开连接</Button>
|
||||||
|
|||||||
Reference in New Issue
Block a user