中继答 504 并说"可能已经执行了,重试前请先确认"——这句话诚实的前提是它保持为真。 批次是串行处理的,所以满邮箱里第八个信封可以在它被签名之后 8 个上游超时才轮到; 那时手机早已被告知失败、用户可能已经重试。再执行它,就把一次诚实的"未知"变成了 一条重复发出的短信。 现在 agent 在动任何东西之前先比对自己的截止时间(策略里的 requestTtlSeconds), 超龄一律以 expired 拒绝并落审计。判定放在新鲜度检查之后、策略判定之前:它比签名 更便宜,而且一个过期请求无论路径是否在白名单里都不该执行。 为了让这条规则**可达**,策略新增一条断言:TTL 不得大于时钟偏移窗口。否则新鲜度 检查会先触发,这条规则永远不会被执行——而"永不触发的规则"是出事时才会被发现坏了 的那种东西。 单个上游调用的超时也移进了策略(agentUpstreamTimeoutSeconds)。"中继期限必须长于 一次上游调用"这个比较,与二进制里的常量是比不了的;现在两侧从同一个文件读,策略 加载时就能拒绝一组自相矛盾的数值。原来那个 20s 常量保留为默认值的来源,不再硬编码。 测试两项:一个请求在期限内被转发、一个超龄请求被拒且**没有被转发**(计数按路径, 否则首次转发前的 /api/auth/status 探测会让断言以错误的理由通过);另一项断言策略 本身满足上述两个关系。变异验证:关掉判定,第一项失败。 Go 全包、Node 119 项、Swift 101 项通过。
662 lines
24 KiB
Go
662 lines
24 KiB
Go
package runtime
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"log/slog"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"net/url"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"fun.chick/msa-agent/internal/policy"
|
|
"fun.chick/msa-agent/internal/wire"
|
|
)
|
|
|
|
// The tests below drive the executor against a real local HTTP server, because the property under
|
|
// test is "the request that reaches SimAdmin is the one the policy allowed". A fake transport would
|
|
// only prove the executor calls an interface.
|
|
|
|
func testPolicy(t *testing.T) *policy.Policy {
|
|
t.Helper()
|
|
pol, err := policy.Load()
|
|
if err != nil {
|
|
t.Fatalf("policy.Load: %v", err)
|
|
}
|
|
return pol
|
|
}
|
|
|
|
// localServer starts a loopback server and returns its port. The executor is pointed at the port
|
|
// only; the server's own host is never handed to it.
|
|
func localServer(t *testing.T, handler http.Handler) int {
|
|
t.Helper()
|
|
server := httptest.NewServer(handler)
|
|
t.Cleanup(server.Close)
|
|
parsed, err := url.Parse(server.URL)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
port, err := strconv.Atoi(parsed.Port())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return port
|
|
}
|
|
|
|
type fixture struct {
|
|
executor *Executor
|
|
now time.Time
|
|
mu sync.Mutex
|
|
}
|
|
|
|
func (f *fixture) clock() time.Time {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return f.now
|
|
}
|
|
|
|
func (f *fixture) advance(d time.Duration) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
f.now = f.now.Add(d)
|
|
}
|
|
|
|
func newFixture(t *testing.T, handler http.Handler) *fixture {
|
|
t.Helper()
|
|
f := &fixture{now: time.Unix(1_800_000_000, 0).UTC()}
|
|
executor, err := NewExecutor(testPolicy(t), Config{
|
|
Port: localServer(t, handler),
|
|
Timeout: 5 * time.Second,
|
|
Audit: NullAudit(),
|
|
Clock: f.clock,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("NewExecutor: %v", err)
|
|
}
|
|
f.executor = executor
|
|
return f
|
|
}
|
|
|
|
func envelopeAt(node, req string, ts uint64) wire.Envelope {
|
|
return wire.Envelope{
|
|
V: wire.Version, Node: node, Req: req, App: "ap_test0000", TS: ts,
|
|
Alg: wire.Suite, EK: make([]byte, 32), P: []byte{1},
|
|
}
|
|
}
|
|
|
|
func request(method, path string, body []byte) wire.InnerRequest {
|
|
return wire.InnerRequest{Method: method, Path: path, Body: body}
|
|
}
|
|
|
|
func TestReadOnTheAllowlistIsForwarded(t *testing.T) {
|
|
var seen http.Request
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
seen = *r
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{"ok":true}`))
|
|
}))
|
|
result, refusal := f.executor.Execute(context.Background(),
|
|
envelopeAt("nd_test0001", "11111111-1111-4111-8111-111111111111", uint64(f.now.Unix())),
|
|
request("GET", "/api/health", nil))
|
|
if refusal != nil {
|
|
t.Fatalf("read was refused: %v", refusal)
|
|
}
|
|
if result.Status != 200 || string(result.Body) != `{"ok":true}` {
|
|
t.Fatalf("got %d %q", result.Status, result.Body)
|
|
}
|
|
if seen.Method != "GET" || seen.URL.Path != "/api/health" {
|
|
t.Fatalf("upstream saw %s %s", seen.Method, seen.URL.Path)
|
|
}
|
|
if seen.Host != "127.0.0.1" && !strings.HasPrefix(seen.Host, "127.0.0.1:") {
|
|
t.Fatalf("upstream saw host %q, want the loopback literal", seen.Host)
|
|
}
|
|
}
|
|
|
|
func TestTheSameRequestIDIsRejectedTheSecondTime(t *testing.T) {
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.WriteHeader(200)
|
|
}))
|
|
envelope := envelopeAt("nd_test0001", "22222222-2222-4222-8222-222222222222", uint64(f.now.Unix()))
|
|
if _, refusal := f.executor.Execute(context.Background(), envelope, request("GET", "/api/health", nil)); refusal != nil {
|
|
t.Fatalf("first call refused: %v", refusal)
|
|
}
|
|
_, refusal := f.executor.Execute(context.Background(), envelope, request("GET", "/api/health", nil))
|
|
if refusal == nil || refusal.Class != ClassReplay {
|
|
t.Fatalf("second call: got %v, want %s", refusal, ClassReplay)
|
|
}
|
|
}
|
|
|
|
// A denied request must still burn its id. If it did not, an attacker holding one signed request
|
|
// could walk it down the whole allowlist by changing only the path.
|
|
func TestADeniedRequestStillBurnsItsID(t *testing.T) {
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
t.Error("upstream must not be called for a path outside the policy")
|
|
}))
|
|
envelope := envelopeAt("nd_test0001", "33333333-3333-4333-8333-333333333333", uint64(f.now.Unix()))
|
|
if _, refusal := f.executor.Execute(context.Background(), envelope, request("GET", "/api/secrets", nil)); refusal == nil || refusal.Class != ClassPolicy {
|
|
t.Fatalf("got %v, want %s", refusal, ClassPolicy)
|
|
}
|
|
_, refusal := f.executor.Execute(context.Background(), envelope, request("GET", "/api/health", nil))
|
|
if refusal == nil || refusal.Class != ClassReplay {
|
|
t.Fatalf("reuse after a denial: got %v, want %s", refusal, ClassReplay)
|
|
}
|
|
}
|
|
|
|
func TestAClockedRequestIsClassifiedAsExpiredNotReplay(t *testing.T) {
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.WriteHeader(200)
|
|
}))
|
|
pol := testPolicy(t)
|
|
stale := uint64(f.now.Unix()) - uint64(pol.Limits.ClockSkewSeconds) - 1
|
|
_, refusal := f.executor.Execute(context.Background(),
|
|
envelopeAt("nd_test0001", "44444444-4444-4444-8444-444444444444", stale),
|
|
request("GET", "/api/health", nil))
|
|
if refusal == nil || refusal.Class != ClassExpired {
|
|
t.Fatalf("got %v, want %s", refusal, ClassExpired)
|
|
}
|
|
future := uint64(f.now.Unix()) + uint64(pol.Limits.ClockSkewSeconds) + 1
|
|
_, refusal = f.executor.Execute(context.Background(),
|
|
envelopeAt("nd_test0001", "55555555-5555-4555-8555-555555555555", future),
|
|
request("GET", "/api/health", nil))
|
|
if refusal == nil || refusal.Class != ClassExpired {
|
|
t.Fatalf("clock ahead: got %v, want %s", refusal, ClassExpired)
|
|
}
|
|
}
|
|
|
|
func TestAuthPathsAreUnreachable(t *testing.T) {
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
// The agent's own status probe is allowed here (and answers "unprotected", so no login
|
|
// happens); what must never be dialed is a login on behalf of a phone.
|
|
if r.URL.Path != "/api/auth/status" {
|
|
t.Error("the login endpoint must never be dialed on behalf of a phone")
|
|
}
|
|
w.WriteHeader(200)
|
|
}))
|
|
for index, path := range []string{"/api/auth/login", "/api/login", "/api/logout"} {
|
|
_, refusal := f.executor.Execute(context.Background(),
|
|
envelopeAt("nd_test0001", newReqID(index), uint64(f.now.Unix())),
|
|
request("POST", path, []byte(`{}`)))
|
|
if refusal == nil || refusal.Class != ClassPolicy {
|
|
t.Fatalf("%s: got %v, want %s", path, refusal, ClassPolicy)
|
|
}
|
|
}
|
|
}
|
|
|
|
// The whole point of the intent: a relay that holds a confirmation for one write cannot spend it on
|
|
// another, even when the two have identical bodies.
|
|
func TestADangerousWriteNeedsAMatchingIntent(t *testing.T) {
|
|
calls := 0
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
// The auth probe reaches the local service too; only the intended path is counted.
|
|
if r.URL.Path == "/api/sms/send" {
|
|
calls++
|
|
}
|
|
w.WriteHeader(200)
|
|
w.Write([]byte(`{"sent":true}`))
|
|
}))
|
|
body := []byte(`{"phone_number":"+8613800000000","content":"hello"}`)
|
|
|
|
if _, refusal := f.executor.Execute(context.Background(),
|
|
envelopeAt("nd_test0001", "77777777-7777-4777-8777-777777777777", uint64(f.now.Unix())),
|
|
request("POST", "/api/sms/send", body)); refusal == nil || refusal.Class != ClassNoConfirm {
|
|
t.Fatalf("without an intent: got %v, want %s", refusal, ClassNoConfirm)
|
|
}
|
|
|
|
right, err := wire.IntentDigest("POST", "/api/sms/send", body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
other, err := wire.IntentDigest("POST", "/api/sms/batch-delete", body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, refusal := f.executor.Execute(context.Background(),
|
|
envelopeAt("nd_test0001", "88888888-8888-4888-8888-888888888888", uint64(f.now.Unix())),
|
|
wire.InnerRequest{Method: "POST", Path: "/api/sms/send", Body: body, Intent: other}); refusal == nil || refusal.Class != ClassBadConfirm {
|
|
t.Fatalf("borrowed intent: got %v, want %s", refusal, ClassBadConfirm)
|
|
}
|
|
|
|
// Lower-cased method and reordered keys must still match: the intent covers the meaning of the
|
|
// request, not the spelling the phone happened to use.
|
|
reordered := []byte(`{"content":"hello","phone_number":"+8613800000000"}`)
|
|
recomputed, err := wire.IntentDigest("post", "/api/sms/send", reordered)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if string(recomputed) != string(right) {
|
|
t.Fatal("intent digest depends on formatting, which it must not")
|
|
}
|
|
|
|
result, refusal := f.executor.Execute(context.Background(),
|
|
envelopeAt("nd_test0001", "99999999-9999-4999-8999-999999999999", uint64(f.now.Unix())),
|
|
wire.InnerRequest{Method: "POST", Path: "/api/sms/send", Body: body, Intent: right})
|
|
if refusal != nil {
|
|
t.Fatalf("with a matching intent: %v", refusal)
|
|
}
|
|
if result.Status != 200 {
|
|
t.Fatalf("status %d", result.Status)
|
|
}
|
|
if calls != 1 {
|
|
t.Fatalf("upstream called %d times, want 1", calls)
|
|
}
|
|
}
|
|
|
|
// Intent must bind the body too. A confirmation for "send to 13800000000" is not consent to send
|
|
// to 13800000001, and the digest is the only thing that says so.
|
|
func TestAnIntentDoesNotCoverAnEditedBody(t *testing.T) {
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
t.Error("an edited body must not reach upstream")
|
|
}))
|
|
intent, err := wire.IntentDigest("POST", "/api/sms/send", []byte(`{"phone_number":"+8613800000000","content":"hi"}`))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, refusal := f.executor.Execute(context.Background(),
|
|
envelopeAt("nd_test0001", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", uint64(f.now.Unix())),
|
|
wire.InnerRequest{Method: "POST", Path: "/api/sms/send",
|
|
Body: []byte(`{"phone_number":"+8613800000001","content":"hi"}`), Intent: intent})
|
|
if refusal == nil || refusal.Class != ClassBadConfirm {
|
|
t.Fatalf("got %v, want %s", refusal, ClassBadConfirm)
|
|
}
|
|
}
|
|
|
|
func TestDangerousWritesAreRateLimited(t *testing.T) {
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.WriteHeader(200)
|
|
}))
|
|
body := []byte(`{"phone_number":"+8613800000000","content":"hello"}`)
|
|
intent, err := wire.IntentDigest("POST", "/api/sms/send", body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
limit := testPolicy(t).Limits.DangerousWritesPerMinute
|
|
allowed := 0
|
|
for i := 0; i < limit+2; i++ {
|
|
inner := wire.InnerRequest{Method: "POST", Path: "/api/sms/send", Body: body, Intent: intent}
|
|
envelope := envelopeAt("nd_test0001", newReqID(i), uint64(f.now.Unix()))
|
|
if _, refusal := f.executor.Execute(context.Background(), envelope, inner); refusal == nil {
|
|
allowed++
|
|
} else if refusal.Class != ClassRateLimited {
|
|
t.Fatalf("attempt %d refused as %s", i, refusal.Class)
|
|
}
|
|
}
|
|
if allowed != limit {
|
|
t.Fatalf("allowed %d dangerous writes, want exactly %d", allowed, limit)
|
|
}
|
|
// The window slides, so a write is admitted exactly when its predecessor leaves the window — not
|
|
// all at once on a calendar boundary. These writes all happened at f.now, so the whole budget
|
|
// frees one minute after that instant.
|
|
f.advance(time.Minute)
|
|
envelope := envelopeAt("nd_test0001", "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", uint64(f.now.Unix()))
|
|
if _, refusal := f.executor.Execute(context.Background(), envelope,
|
|
wire.InnerRequest{Method: "POST", Path: "/api/sms/send", Body: body, Intent: intent}); refusal != nil {
|
|
t.Fatalf("after the window: %v", refusal)
|
|
}
|
|
}
|
|
|
|
// The property, stated as the defect it replaces.
|
|
//
|
|
// A fixed window — a counter reset on a calendar minute — admits 2N writes inside a few milliseconds
|
|
// by spending one budget just before its boundary and the next just after. The relay's per-device
|
|
// budget had exactly that flaw (see relay/src/budget.js), so the two limiters are now the same
|
|
// algorithm; this test is what makes the node's half of that claim checkable rather than asserted.
|
|
func TestDangerousWriteLimitDoesNotResetOnACalendarBoundary(t *testing.T) {
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.WriteHeader(200)
|
|
}))
|
|
body := []byte(`{"phone_number":"+8613800000000","content":"hello"}`)
|
|
intent, err := wire.IntentDigest("POST", "/api/sms/send", body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
limit := testPolicy(t).Limits.DangerousWritesPerMinute
|
|
|
|
attempt := 0
|
|
send := func() bool {
|
|
inner := wire.InnerRequest{Method: "POST", Path: "/api/sms/send", Body: body, Intent: intent}
|
|
envelope := envelopeAt("nd_test0001", newReqID(attempt), uint64(f.now.Unix()))
|
|
attempt++
|
|
_, refusal := f.executor.Execute(context.Background(), envelope, inner)
|
|
return refusal == nil
|
|
}
|
|
|
|
// Park the clock one second before a calendar-minute boundary, then spend the whole budget there.
|
|
// The fixture starts at a round minute, so 59 seconds lands on :59 of that minute.
|
|
f.advance(59 * time.Second)
|
|
for i := 0; i < limit; i++ {
|
|
if !send() {
|
|
t.Fatalf("write %d inside the budget was refused", i)
|
|
}
|
|
}
|
|
|
|
// Cross the boundary. A fixed window has just reset its counter and admits another full budget —
|
|
// which is the defect this test exists for. A sliding one admits nothing: the writes are two
|
|
// seconds old, and the window is 60.
|
|
f.advance(2 * time.Second)
|
|
admitted := 0
|
|
for i := 0; i < limit; i++ {
|
|
if send() {
|
|
admitted++
|
|
}
|
|
}
|
|
if admitted != 0 {
|
|
t.Fatalf("%d writes were admitted across the calendar boundary; the window is not sliding", admitted)
|
|
}
|
|
|
|
// And the budget frees one write at a time as the earlier ones age out, not all at once: 58 more
|
|
// seconds puts the writes from :59 past the window, so the full budget is available again.
|
|
f.advance(58 * time.Second)
|
|
admitted = 0
|
|
for i := 0; i < limit; i++ {
|
|
if send() {
|
|
admitted++
|
|
}
|
|
}
|
|
if admitted != limit {
|
|
t.Fatalf("after the window, %d writes were admitted, want exactly %d", admitted, limit)
|
|
}
|
|
}
|
|
|
|
func newReqID(i int) string {
|
|
return "cccccccc-cccc-4ccc-8ccc-" + strings.Repeat("0", 11) + strconv.Itoa(i)
|
|
}
|
|
|
|
// An oversized answer is refused, not truncated. A truncated body is a broken body: JSON cut at an
|
|
// arbitrary offset parses as nothing, and the user would see a decoding error that blames the app
|
|
// for a node that answered with 3 MB.
|
|
func TestOversizedResponsesAreRefusedNotTruncated(t *testing.T) {
|
|
limit := testPolicy(t).Limits.MaxResponseBytes
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.Write([]byte(strings.Repeat("x", limit+1024)))
|
|
}))
|
|
_, refusal := f.executor.Execute(context.Background(),
|
|
envelopeAt("nd_test0001", "dddddddd-dddd-4ddd-8ddd-dddddddddddd", uint64(f.now.Unix())),
|
|
request("GET", "/api/health", nil))
|
|
if refusal == nil || refusal.Class != ClassTooLarge {
|
|
t.Fatalf("got %v, want %s", refusal, ClassTooLarge)
|
|
}
|
|
}
|
|
|
|
// The boundary itself must pass: a response of exactly the limit is a normal response.
|
|
func TestAResponseExactlyAtTheLimitIsForwarded(t *testing.T) {
|
|
limit := testPolicy(t).Limits.MaxResponseBytes
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.Write([]byte(strings.Repeat("x", limit)))
|
|
}))
|
|
result, refusal := f.executor.Execute(context.Background(),
|
|
envelopeAt("nd_test0001", "d1d1d1d1-d1d1-41d1-81d1-d1d1d1d1d1d1", uint64(f.now.Unix())),
|
|
request("GET", "/api/health", nil))
|
|
if refusal != nil {
|
|
t.Fatalf("refused: %v", refusal)
|
|
}
|
|
if len(result.Body) != limit {
|
|
t.Fatalf("body is %d bytes, want %d", len(result.Body), limit)
|
|
}
|
|
}
|
|
|
|
func TestAnUnreachableLocalServiceIsANodeAuthError(t *testing.T) {
|
|
pol := testPolicy(t)
|
|
// A port with nothing on it: the authentication probe fails first, which is a node problem
|
|
// rather than a policy problem. It is classified as node_auth_failed rather than
|
|
// upstream_error because nothing of the phone's command was ever attempted, and the phone's
|
|
// remedy ("check that SimAdmin is running on this node") is the operator's, not the user's.
|
|
executor, err := NewExecutor(pol, Config{Port: 1, Timeout: 500 * time.Millisecond, Audit: NullAudit()})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, refusal := executor.Execute(context.Background(),
|
|
envelopeAt("nd_test0001", "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee", uint64(time.Now().Unix())),
|
|
request("GET", "/api/health", nil))
|
|
if refusal == nil || refusal.Class != ClassNodeAuth {
|
|
t.Fatalf("got %v, want %s", refusal, ClassNodeAuth)
|
|
}
|
|
}
|
|
|
|
func TestReplayCacheBoundsAndTTL(t *testing.T) {
|
|
cache, err := NewReplayCache(3, time.Minute)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
base := time.Unix(1_800_000_000, 0)
|
|
for i := 0; i < 3; i++ {
|
|
if cache.Seen(strconv.Itoa(i), base) {
|
|
t.Fatalf("id %d reported as seen", i)
|
|
}
|
|
}
|
|
if !cache.Seen("0", base) {
|
|
t.Fatal("id 0 was evicted before the cache was full")
|
|
}
|
|
// Past the ttl every entry is gone, so the bound and the window are independent of each other.
|
|
later := base.Add(2 * time.Minute)
|
|
if cache.Len(later) != 0 {
|
|
t.Fatalf("cache holds %d entries past its ttl", cache.Len(later))
|
|
}
|
|
if cache.Seen("0", later) {
|
|
t.Fatal("an expired id is still reported as seen")
|
|
}
|
|
}
|
|
|
|
func TestFreshnessWindow(t *testing.T) {
|
|
if err := Freshness(1000, 1000, 120); err != nil {
|
|
t.Fatalf("same timestamp: %v", err)
|
|
}
|
|
if err := Freshness(1000, 1121, 120); !errors.Is(err, ErrExpired) {
|
|
t.Fatalf("121s old: %v", err)
|
|
}
|
|
// Symmetric: a request stamped in the future is as suspicious as an old one, and the window is
|
|
// a bound on |now - ts|, not on age.
|
|
if err := Freshness(1121, 1000, 120); !errors.Is(err, ErrExpired) {
|
|
t.Fatalf("121s ahead: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestAuditWritesOneJSONLinePerRequest(t *testing.T) {
|
|
path := filepath.Join(t.TempDir(), "nested", "audit.log")
|
|
audit, err := OpenAudit(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer audit.Close()
|
|
stamp := time.Unix(1_800_000_000, 0).UTC()
|
|
if err := audit.Write(Entry{At: stamp, Node: "nd_1", Req: "r1", Method: "GET", Path: "/api/health", Status: 200, Outcome: "ok"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := audit.Write(Entry{At: stamp, Req: "r2", Outcome: ClassPolicy}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
raw, err := os.ReadFile(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
lines := strings.Split(strings.TrimSpace(string(raw)), "\n")
|
|
if len(lines) != 2 {
|
|
t.Fatalf("got %d lines, want 2", len(lines))
|
|
}
|
|
var first Entry
|
|
if err := json.Unmarshal([]byte(lines[0]), &first); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if first.Node != "nd_1" || first.Status != 200 || !first.At.Equal(stamp) {
|
|
t.Fatalf("first entry round-tripped as %+v", first)
|
|
}
|
|
info, err := os.Stat(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if info.Mode().Perm() != 0o600 {
|
|
t.Fatalf("audit file mode is %v, want 0600", info.Mode().Perm())
|
|
}
|
|
}
|
|
|
|
// A second run over the same log must append, not truncate: an agent restart is normal and must not
|
|
// erase the explanation for what happened before it.
|
|
func TestAuditAppendsAcrossRestarts(t *testing.T) {
|
|
path := filepath.Join(t.TempDir(), "audit.log")
|
|
first, err := OpenAudit(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := first.Write(Entry{Req: "r1", Outcome: "ok"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := first.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
second, err := OpenAudit(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer second.Close()
|
|
if err := second.Write(Entry{Req: "r2", Outcome: ClassReplay}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
raw, err := os.ReadFile(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got := strings.Count(string(raw), "\n"); got != 2 {
|
|
t.Fatalf("log has %d lines, want 2", got)
|
|
}
|
|
}
|
|
|
|
func TestNullAuditAcceptsWritesWithoutAFile(t *testing.T) {
|
|
audit := NullAudit()
|
|
if err := audit.Write(Entry{Req: "r"}); err != nil {
|
|
t.Fatalf("null audit returned %v", err)
|
|
}
|
|
if err := audit.Close(); err != nil {
|
|
t.Fatalf("null audit close returned %v", err)
|
|
}
|
|
}
|
|
|
|
func TestExecutorRefusesToStartWithoutAPolicyOrPort(t *testing.T) {
|
|
if _, err := NewExecutor(nil, Config{Port: 8788}); err == nil {
|
|
t.Fatal("a nil policy was accepted")
|
|
}
|
|
if _, err := NewExecutor(testPolicy(t), Config{Port: 0}); err == nil {
|
|
t.Fatal("port 0 was accepted")
|
|
}
|
|
}
|
|
|
|
// A failed audit write must be visible somewhere. The entry cannot fail the request — the audit log is
|
|
// evidence, not a control, and a full disk must not become a remote denial of service — but silently
|
|
// losing entries is the failure mode where an operator believes they have a record and does not.
|
|
func TestAFailedAuditWriteIsReported(t *testing.T) {
|
|
directory := t.TempDir()
|
|
path := filepath.Join(directory, "audit.log")
|
|
audit, err := OpenAudit(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// Replace the sink with one that fails, which is what a full disk looks like from here. Reaching
|
|
// into the struct is deliberate: the alternative is filling a filesystem in a test.
|
|
audit.mu.Lock()
|
|
audit.sink = nil
|
|
audit.mu.Unlock()
|
|
|
|
var captured bytes.Buffer
|
|
previous := slog.Default()
|
|
slog.SetDefault(slog.New(slog.NewJSONHandler(&captured, &slog.HandlerOptions{Level: slog.LevelDebug})))
|
|
defer slog.SetDefault(previous)
|
|
|
|
// Close() drops the sink, so a write after it returns nil rather than an error — that is the
|
|
// documented shutdown case. Force the error path by giving it a closed file instead.
|
|
file, err := os.OpenFile(filepath.Join(directory, "closed.log"), os.O_CREATE|os.O_WRONLY, 0o600)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := file.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
audit.mu.Lock()
|
|
audit.sink = file
|
|
audit.mu.Unlock()
|
|
|
|
if err := audit.Write(Entry{Node: "nd_x", Req: "r1", Outcome: "ok"}); err == nil {
|
|
t.Fatal("writing to a closed file reported success")
|
|
}
|
|
// And the executor's own path reports it rather than swallowing it.
|
|
exec, err := NewExecutor(testPolicy(t), Config{Port: 3000, Audit: audit})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
exec.record(wire.Envelope{Node: "nd_x", Req: "r2", App: "ap_x"}, "GET", "/api/health", 200, "ok")
|
|
if !strings.Contains(captured.String(), "audit entry could not be written") {
|
|
t.Fatalf("a failed audit write was not reported; log was: %s", captured.String())
|
|
}
|
|
}
|
|
|
|
// A request whose phone has already given up must not be executed.
|
|
//
|
|
// The relay answers 504 with "it may or may not have executed it; check the node before retrying",
|
|
// which is honest — but only while it stays true. A batch is processed serially, so the eighth
|
|
// envelope of a full mailbox can begin up to eight upstream timeouts after it was signed. By then the
|
|
// phone has been told the request failed and the user may have retried; executing it turns an honest
|
|
// "unknown" into a duplicate SMS.
|
|
func TestARequestPastItsDeadlineIsNotExecuted(t *testing.T) {
|
|
// Counted by path: the agent probes /api/auth/status before its first forward, and counting every
|
|
// request would make the first assertion pass for the wrong reason.
|
|
executed := 0
|
|
f := newFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path == "/api/sms/send" {
|
|
executed++
|
|
}
|
|
w.WriteHeader(200)
|
|
}))
|
|
pol := testPolicy(t)
|
|
body := []byte(`{"phone_number":"+8613800000000","content":"hello"}`)
|
|
intent, err := wire.IntentDigest("POST", "/api/sms/send", body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
inner := wire.InnerRequest{Method: "POST", Path: "/api/sms/send", Body: body, Intent: intent}
|
|
|
|
// One second inside the deadline: executed, because the phone is still waiting.
|
|
envelope := envelopeAt("nd_test0001", newReqID(1), uint64(f.now.Unix()))
|
|
if _, refusal := f.executor.Execute(context.Background(), envelope, inner); refusal != nil {
|
|
t.Fatalf("a request inside its deadline was refused: %v", refusal)
|
|
}
|
|
if executed != 1 {
|
|
t.Fatalf("expected one execution, got %d", executed)
|
|
}
|
|
|
|
// One second past it: refused, and nothing is forwarded.
|
|
f.advance(time.Duration(pol.Limits.RequestTTLSeconds+1) * time.Second)
|
|
late := envelopeAt("nd_test0001", newReqID(2), uint64(f.now.Unix()-int64(pol.Limits.RequestTTLSeconds)-1))
|
|
_, refusal := f.executor.Execute(context.Background(), late, inner)
|
|
if refusal == nil {
|
|
t.Fatal("a request past its deadline was executed; the phone was told it had failed")
|
|
}
|
|
if refusal.Class != ClassExpired {
|
|
t.Fatalf("refused as %s, want %s", refusal.Class, ClassExpired)
|
|
}
|
|
if executed != 1 {
|
|
t.Fatalf("the late request was forwarded anyway (%d executions)", executed)
|
|
}
|
|
}
|
|
|
|
// The deadline check is only reachable because the policy keeps the TTL inside the clock-skew window.
|
|
// If it were not, the freshness check would fire first and the deadline rule could never be reached —
|
|
// a rule that never fires is one nobody discovers is broken until an incident.
|
|
func TestThePolicyKeepsTheDeadlineInsideTheFreshnessWindow(t *testing.T) {
|
|
pol := testPolicy(t)
|
|
if pol.Limits.RequestTTLSeconds > pol.Limits.ClockSkewSeconds {
|
|
t.Fatalf("requestTtlSeconds (%d) exceeds clockSkewSeconds (%d); the deadline check is unreachable",
|
|
pol.Limits.RequestTTLSeconds, pol.Limits.ClockSkewSeconds)
|
|
}
|
|
// And the relay's deadline is the longer bound, or the phone gives up before the node does.
|
|
if pol.Limits.RequestTTLSeconds <= pol.Limits.AgentUpstreamTimeoutSeconds {
|
|
t.Fatalf("the relay deadline (%ds) is not longer than one upstream call (%ds)",
|
|
pol.Limits.RequestTTLSeconds, pol.Limits.AgentUpstreamTimeoutSeconds)
|
|
}
|
|
}
|