| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "context" |
| 6 | "errors" |
| 7 | "net" |
| 8 | "net/http" |
| 9 | "net/url" |
| 10 | "os" |
| 11 | "os/exec" |
| 12 | "path/filepath" |
| 13 | "strconv" |
| 14 | "strings" |
| 15 | "sync" |
| 16 | "syscall" |
| 17 | "testing" |
| 18 | "time" |
| 19 | ) |
| 20 | |
| 21 | // TestMain lets TestSIGTERMDrainsQueuedSigninMail run this binary as |
| 22 | // the real process: with CALLS_TEST_RUN_MAIN=1 it is main() and nothing |
| 23 | // else, so os.Args are the activation flags the test passes. |
| 24 | func TestMain(m *testing.M) { |
| 25 | if os.Getenv("CALLS_TEST_RUN_MAIN") == "1" { |
| 26 | main() |
| 27 | os.Exit(0) |
| 28 | } |
| 29 | os.Exit(m.Run()) |
| 30 | } |
| 31 | |
| 32 | // TestSIGTERMDrainsQueuedSigninMail: SMTP runs after the response, in |
| 33 | // the sign-in mail queue, so Server.Shutdown no longer waits for it. |
| 34 | // A link that was already answered ?sent=1 must still go out when the |
| 35 | // process is stopped: main has to wait for the queue's bounded drain |
| 36 | // before it closes the database and exits (Astra on oficina/meet #25). |
| 37 | // |
| 38 | // The relay holds the message at DATA until after SIGTERM has arrived, |
| 39 | // then accepts it well inside the drain window. The process must exit |
| 40 | // cleanly, and only after the relay has the message. |
| 41 | func TestSIGTERMDrainsQueuedSigninMail(t *testing.T) { |
| 42 | relay := newHeldRelay(t) |
| 43 | p := startCalls(t, relay.port) |
| 44 | |
| 45 | // force=1: an emailed link without asking anyone whether |
| 46 | // example.test runs Keymail. |
| 47 | form := url.Values{"address": {"paul@example.test"}, "force": {"1"}} |
| 48 | req, _ := http.NewRequest(http.MethodPost, p.origin+"/signin", strings.NewReader(form.Encode())) |
| 49 | req.Header.Set("Content-Type", "application/x-www-form-urlencoded") |
| 50 | req.Header.Set("Origin", p.origin) |
| 51 | req.Header.Set("Sec-Fetch-Site", "same-origin") |
| 52 | res, err := p.client.Do(req) |
| 53 | if err != nil { |
| 54 | t.Fatal(err) |
| 55 | } |
| 56 | res.Body.Close() |
| 57 | if res.StatusCode != http.StatusSeeOther || !strings.Contains(res.Header.Get("Location"), "sent=1") { |
| 58 | t.Fatalf("POST /signin: %d %q, want 303 ?sent=1", res.StatusCode, res.Header.Get("Location")) |
| 59 | } |
| 60 | |
| 61 | select { |
| 62 | case <-relay.atData: |
| 63 | case <-time.After(10 * time.Second): |
| 64 | t.Fatal("the queued link never reached the relay") |
| 65 | } |
| 66 | if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil { |
| 67 | t.Fatal(err) |
| 68 | } |
| 69 | // Long enough that a process which does not wait has gone (Astra's |
| 70 | // probe saw it exit in 4 ms), short against the 30 s drain window. |
| 71 | select { |
| 72 | case err := <-p.exited: |
| 73 | t.Fatalf("process exited (%v) while its acknowledged sign-in link was still with the relay", err) |
| 74 | case <-time.After(500 * time.Millisecond): |
| 75 | } |
| 76 | relay.release() |
| 77 | p.waitClean(t) |
| 78 | if !relay.accepted() { |
| 79 | t.Fatal("process exited without the relay accepting the sign-in link") |
| 80 | } |
| 81 | } |
| 82 | |
| 83 | // TestSIGTERMDeliversAnInFlightSigninsLink: Server.Shutdown lets a |
| 84 | // request it had already accepted finish, so that request's link must |
| 85 | // still be queued and sent, not refused by a queue that SIGTERM closed |
| 86 | // while the handler was still running (Astra on oficina/meet #25, round |
| 87 | // 2: the request answered ?sent=1 and no mail went out). |
| 88 | // |
| 89 | // The POST's headers and all but the last byte of its body are sent |
| 90 | // before SIGTERM, so the handler is in flight, still reading the form. |
| 91 | // The last byte follows once the process has logged its shutdown. |
| 92 | func TestSIGTERMDeliversAnInFlightSigninsLink(t *testing.T) { |
| 93 | relay := newHeldRelay(t) |
| 94 | relay.release() // an ordinary relay: this test is about intake |
| 95 | p := startCalls(t, relay.port) |
| 96 | |
| 97 | body := url.Values{"address": {"paul@example.test"}, "force": {"1"}}.Encode() |
| 98 | conn, err := net.Dial("tcp", p.addr) |
| 99 | if err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | defer conn.Close() |
| 103 | head := "POST /signin HTTP/1.1\r\nHost: " + p.addr + "\r\nOrigin: " + p.origin + |
| 104 | "\r\nSec-Fetch-Site: same-origin\r\nContent-Type: application/x-www-form-urlencoded\r\n" + |
| 105 | "Content-Length: " + strconv.Itoa(len(body)) + "\r\nConnection: close\r\n\r\n" |
| 106 | if _, err := conn.Write([]byte(head + body[:len(body)-1])); err != nil { |
| 107 | t.Fatal(err) |
| 108 | } |
| 109 | // Let the server start the handler on what it has. |
| 110 | time.Sleep(200 * time.Millisecond) |
| 111 | |
| 112 | if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil { |
| 113 | t.Fatal(err) |
| 114 | } |
| 115 | until := time.Now().Add(10 * time.Second) |
| 116 | for !strings.Contains(p.out.String(), "rastrillo: shutting down") { |
| 117 | if time.Now().After(until) { |
| 118 | t.Fatal("the process never began shutting down") |
| 119 | } |
| 120 | time.Sleep(10 * time.Millisecond) |
| 121 | } |
| 122 | time.Sleep(300 * time.Millisecond) |
| 123 | if _, err := conn.Write([]byte(body[len(body)-1:])); err != nil { |
| 124 | t.Fatal(err) |
| 125 | } |
| 126 | _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second)) |
| 127 | res, err := http.ReadResponse(bufio.NewReader(conn), nil) |
| 128 | if err != nil { |
| 129 | t.Fatalf("the in-flight POST got no answer: %v", err) |
| 130 | } |
| 131 | res.Body.Close() |
| 132 | if res.StatusCode != http.StatusSeeOther || !strings.Contains(res.Header.Get("Location"), "sent=1") { |
| 133 | t.Fatalf("in-flight POST /signin: %d %q, want 303 ?sent=1", res.StatusCode, res.Header.Get("Location")) |
| 134 | } |
| 135 | |
| 136 | p.waitClean(t) |
| 137 | if !relay.accepted() { |
| 138 | t.Fatal("the in-flight request was answered ?sent=1, but the process exited without sending its link") |
| 139 | } |
| 140 | } |
| 141 | |
| 142 | // TestSIGTERMDrainsMailWhenHTTPShutdownTimesOut: rastrillo.Serve |
| 143 | // returns an error when its ten-second Server.Shutdown deadline passes |
| 144 | // with a request still running, and main used to os.Exit(1) on it, |
| 145 | // skipping the deferred drain. An unrelated slow request then cost an |
| 146 | // already-acknowledged link (Astra on oficina/meet #25, round 3). The |
| 147 | // drain, and the database close after it, must run on that path too. |
| 148 | // |
| 149 | // One sign-in is answered ?sent=1 and held at the relay's DATA; a second |
| 150 | // POST never finishes its body, so HTTP shutdown times out. The relay |
| 151 | // recovers after the HTTP deadline but well inside the mail drain |
| 152 | // window. The process must still be alive then, deliver the link, and |
| 153 | // exit 1 for the failed shutdown. |
| 154 | func TestSIGTERMDrainsMailWhenHTTPShutdownTimesOut(t *testing.T) { |
| 155 | if testing.Short() { |
| 156 | t.Skip("waits out rastrillo's ten-second shutdown deadline") |
| 157 | } |
| 158 | relay := newHeldRelay(t) |
| 159 | p := startCalls(t, relay.port) |
| 160 | |
| 161 | form := url.Values{"address": {"paul@example.test"}, "force": {"1"}} |
| 162 | req, _ := http.NewRequest(http.MethodPost, p.origin+"/signin", strings.NewReader(form.Encode())) |
| 163 | req.Header.Set("Content-Type", "application/x-www-form-urlencoded") |
| 164 | req.Header.Set("Origin", p.origin) |
| 165 | req.Header.Set("Sec-Fetch-Site", "same-origin") |
| 166 | res, err := p.client.Do(req) |
| 167 | if err != nil { |
| 168 | t.Fatal(err) |
| 169 | } |
| 170 | res.Body.Close() |
| 171 | if res.StatusCode != http.StatusSeeOther || !strings.Contains(res.Header.Get("Location"), "sent=1") { |
| 172 | t.Fatalf("POST /signin: %d %q, want 303 ?sent=1", res.StatusCode, res.Header.Get("Location")) |
| 173 | } |
| 174 | select { |
| 175 | case <-relay.atData: |
| 176 | case <-time.After(10 * time.Second): |
| 177 | t.Fatal("the queued link never reached the relay") |
| 178 | } |
| 179 | |
| 180 | // A request that never finishes: HTTP shutdown cannot complete. |
| 181 | stalled, err := net.Dial("tcp", p.addr) |
| 182 | if err != nil { |
| 183 | t.Fatal(err) |
| 184 | } |
| 185 | defer stalled.Close() |
| 186 | body := url.Values{"address": {"other@example.test"}, "force": {"1"}}.Encode() |
| 187 | if _, err := stalled.Write([]byte("POST /signin HTTP/1.1\r\nHost: " + p.addr + "\r\nOrigin: " + p.origin + |
| 188 | "\r\nSec-Fetch-Site: same-origin\r\nContent-Type: application/x-www-form-urlencoded\r\n" + |
| 189 | "Content-Length: " + strconv.Itoa(len(body)) + "\r\n\r\n" + body[:len(body)-1])); err != nil { |
| 190 | t.Fatal(err) |
| 191 | } |
| 192 | time.Sleep(200 * time.Millisecond) |
| 193 | |
| 194 | if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil { |
| 195 | t.Fatal(err) |
| 196 | } |
| 197 | signalled := time.Now() |
| 198 | select { |
| 199 | case err := <-p.exited: |
| 200 | t.Fatalf("process exited (%v) %v after SIGTERM, with an acknowledged link still at the relay", err, time.Since(signalled).Round(time.Millisecond)) |
| 201 | case <-time.After(11500 * time.Millisecond): |
| 202 | } |
| 203 | relay.release() |
| 204 | |
| 205 | select { |
| 206 | case err := <-p.exited: |
| 207 | var exit *exec.ExitError |
| 208 | if !errors.As(err, &exit) || exit.ExitCode() != 1 { |
| 209 | t.Fatalf("process exit: %v, want status 1 for the failed HTTP shutdown", err) |
| 210 | } |
| 211 | case <-time.After(20 * time.Second): |
| 212 | t.Fatal("process did not exit after the relay accepted the message") |
| 213 | } |
| 214 | if !relay.accepted() { |
| 215 | t.Fatal("process exited without the relay accepting the acknowledged sign-in link") |
| 216 | } |
| 217 | if !strings.Contains(p.out.String(), "context deadline exceeded") { |
| 218 | t.Fatal("HTTP shutdown did not time out; the test did not exercise the failed-shutdown path") |
| 219 | } |
| 220 | } |
| 221 | |
| 222 | // proc is a Calls process started by startCalls. |
| 223 | type proc struct { |
| 224 | addr, origin string |
| 225 | cmd *exec.Cmd |
| 226 | exited chan error |
| 227 | out *lockedBuffer |
| 228 | client *http.Client |
| 229 | } |
| 230 | |
| 231 | // startCalls runs this test binary as main() in development, mailing |
| 232 | // through the relay on relayPort, and waits until it serves. |
| 233 | func startCalls(t *testing.T, relayPort string) *proc { |
| 234 | t.Helper() |
| 235 | ln, err := net.Listen("tcp", "127.0.0.1:0") |
| 236 | if err != nil { |
| 237 | t.Fatal(err) |
| 238 | } |
| 239 | addr := ln.Addr().String() |
| 240 | ln.Close() |
| 241 | |
| 242 | ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) |
| 243 | t.Cleanup(cancel) |
| 244 | dir := t.TempDir() |
| 245 | cmd := exec.CommandContext(ctx, os.Args[0], "-env", "development", "-addr", addr, "-db", filepath.Join(dir, "calls.db")) |
| 246 | cmd.Dir = dir |
| 247 | for _, e := range os.Environ() { |
| 248 | if strings.HasPrefix(e, "CALLS_") || strings.HasPrefix(e, "CARLOS_") || strings.HasPrefix(e, "HOME_") { |
| 249 | continue |
| 250 | } |
| 251 | cmd.Env = append(cmd.Env, e) |
| 252 | } |
| 253 | cmd.Env = append(cmd.Env, |
| 254 | "CALLS_TEST_RUN_MAIN=1", |
| 255 | "CALLS_SMTP_HOST=127.0.0.1", |
| 256 | "CALLS_SMTP_PORT="+relayPort, |
| 257 | "CALLS_SMTP_FROM=calls@example.test", |
| 258 | "CALLS_ALLOWED_ADDRESSES=paul@example.test", |
| 259 | ) |
| 260 | out := &lockedBuffer{} |
| 261 | cmd.Stdout, cmd.Stderr = out, out |
| 262 | if err := cmd.Start(); err != nil { |
| 263 | t.Fatal(err) |
| 264 | } |
| 265 | p := &proc{addr: addr, origin: "http://" + addr, cmd: cmd, exited: make(chan error, 1), out: out, |
| 266 | client: &http.Client{Timeout: 5 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}} |
| 267 | go func() { p.exited <- cmd.Wait() }() |
| 268 | t.Cleanup(func() { |
| 269 | _ = cmd.Process.Kill() |
| 270 | if t.Failed() { |
| 271 | t.Logf("process output:\n%s", out.String()) |
| 272 | } |
| 273 | }) |
| 274 | |
| 275 | up := time.Now().Add(20 * time.Second) |
| 276 | for { |
| 277 | res, err := p.client.Get(p.origin + "/healthz") |
| 278 | if err == nil { |
| 279 | res.Body.Close() |
| 280 | return p |
| 281 | } |
| 282 | if time.Now().After(up) { |
| 283 | t.Fatalf("server never came up: %v", err) |
| 284 | } |
| 285 | time.Sleep(20 * time.Millisecond) |
| 286 | } |
| 287 | } |
| 288 | |
| 289 | // waitClean waits for the process to exit, and requires it to exit 0. |
| 290 | func (p *proc) waitClean(t *testing.T) { |
| 291 | t.Helper() |
| 292 | select { |
| 293 | case err := <-p.exited: |
| 294 | if err != nil { |
| 295 | t.Fatalf("process exit: %v", err) |
| 296 | } |
| 297 | case <-time.After(20 * time.Second): |
| 298 | t.Fatal("process did not exit") |
| 299 | } |
| 300 | } |
| 301 | |
| 302 | // lockedBuffer is the process's output, readable while it is written. |
| 303 | type lockedBuffer struct { |
| 304 | mu sync.Mutex |
| 305 | b strings.Builder |
| 306 | } |
| 307 | |
| 308 | func (l *lockedBuffer) Write(p []byte) (int, error) { |
| 309 | l.mu.Lock() |
| 310 | defer l.mu.Unlock() |
| 311 | return l.b.Write(p) |
| 312 | } |
| 313 | |
| 314 | func (l *lockedBuffer) String() string { |
| 315 | l.mu.Lock() |
| 316 | defer l.mu.Unlock() |
| 317 | return l.b.String() |
| 318 | } |
| 319 | |
| 320 | // heldRelay is a plaintext SMTP server on loopback (no STARTTLS, no |
| 321 | // AUTH, which mail.SMTP accepts when no credentials are set) that holds |
| 322 | // the first message at DATA until release. |
| 323 | type heldRelay struct { |
| 324 | port string |
| 325 | atData chan struct{} |
| 326 | hold chan struct{} |
| 327 | once sync.Once |
| 328 | mu sync.Mutex |
| 329 | gotMail bool |
| 330 | } |
| 331 | |
| 332 | func newHeldRelay(t *testing.T) *heldRelay { |
| 333 | ln, err := net.Listen("tcp", "127.0.0.1:0") |
| 334 | if err != nil { |
| 335 | t.Fatal(err) |
| 336 | } |
| 337 | t.Cleanup(func() { ln.Close() }) |
| 338 | _, port, _ := net.SplitHostPort(ln.Addr().String()) |
| 339 | r := &heldRelay{port: port, atData: make(chan struct{}), hold: make(chan struct{})} |
| 340 | t.Cleanup(r.release) |
| 341 | go func() { |
| 342 | var signalled sync.Once |
| 343 | for { |
| 344 | c, err := ln.Accept() |
| 345 | if err != nil { |
| 346 | return |
| 347 | } |
| 348 | go r.serve(c, &signalled) |
| 349 | } |
| 350 | }() |
| 351 | return r |
| 352 | } |
| 353 | |
| 354 | func (r *heldRelay) release() { r.once.Do(func() { close(r.hold) }) } |
| 355 | |
| 356 | func (r *heldRelay) accepted() bool { |
| 357 | r.mu.Lock() |
| 358 | defer r.mu.Unlock() |
| 359 | return r.gotMail |
| 360 | } |
| 361 | |
| 362 | func (r *heldRelay) serve(c net.Conn, signalled *sync.Once) { |
| 363 | defer c.Close() |
| 364 | br := bufio.NewReader(c) |
| 365 | say := func(s string) { _, _ = c.Write([]byte(s + "\r\n")) } |
| 366 | say("220 relay.test ESMTP") |
| 367 | for { |
| 368 | line, err := br.ReadString('\n') |
| 369 | if err != nil { |
| 370 | return |
| 371 | } |
| 372 | cmd := strings.ToUpper(strings.TrimSpace(line)) |
| 373 | switch { |
| 374 | case strings.HasPrefix(cmd, "EHLO"): |
| 375 | say("250-relay.test") |
| 376 | say("250 8BITMIME") |
| 377 | case strings.HasPrefix(cmd, "HELO"), strings.HasPrefix(cmd, "MAIL"), strings.HasPrefix(cmd, "RCPT"), strings.HasPrefix(cmd, "RSET"), strings.HasPrefix(cmd, "NOOP"): |
| 378 | say("250 OK") |
| 379 | case cmd == "DATA": |
| 380 | signalled.Do(func() { close(r.atData) }) |
| 381 | <-r.hold |
| 382 | say("354 go ahead") |
| 383 | for { |
| 384 | l, err := br.ReadString('\n') |
| 385 | if err != nil { |
| 386 | return |
| 387 | } |
| 388 | if l == ".\r\n" || l == ".\n" { |
| 389 | break |
| 390 | } |
| 391 | } |
| 392 | r.mu.Lock() |
| 393 | r.gotMail = true |
| 394 | r.mu.Unlock() |
| 395 | say("250 queued") |
| 396 | case cmd == "QUIT": |
| 397 | say("221 bye") |
| 398 | return |
| 399 | default: |
| 400 | say("502 unrecognised") |
| 401 | } |
| 402 | } |
| 403 | } |
| 404 | |
amadan