Compare commits
2
Commits
dd762ce759
...
924d4b7177
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
924d4b7177 | ||
|
|
f3895789d2 |
@@ -22,6 +22,20 @@ files, so merging it also closes most compliance gaps.
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
|
- 2026-09-21: shutdown lifecycle correctness. The process now shuts down through
|
||||||
|
fx instead of `os.Exit`, so every component's `OnStop` runs and buffered
|
||||||
|
reports are flushed to disk on `SIGTERM` — previously a full flush window of
|
||||||
|
telemetry was silently lost on every restart. The `http.Server` is now built
|
||||||
|
before its serving goroutine starts, so shutdown can no longer race or
|
||||||
|
nil-deref it; a listen failure exits non-zero via `fx.Shutdowner`; `reportbuf`
|
||||||
|
`OnStop` is idempotent; and `writeTimeout` now exceeds the chi per-request
|
||||||
|
budget so that budget is actually reachable. Dead `startupTime`, `exitCode`,
|
||||||
|
and `cancelFunc` fields were removed
|
||||||
|
- 2026-09-21: backend HTTP hardening (issue #19): added `ReadHeaderTimeout` and
|
||||||
|
`IdleTimeout` to the server, a `SecurityHeaders` middleware (HSTS, tight CSP,
|
||||||
|
frame/sniff/referrer/permissions headers) registered before CORS, and
|
||||||
|
trusted-proxy client IP resolution honouring `X-Forwarded-For` / `X-Real-IP`
|
||||||
|
only from a `TRUSTED_PROXIES` allowlist (loopback plus RFC1918 by default)
|
||||||
- 2026-08-10: every interactive control now meets the 44x44 CSS px minimum tap
|
- 2026-08-10: every interactive control now meets the 44x44 CSS px minimum tap
|
||||||
target (`.pin-btn`, `#interval-select`, the debug-log label and, on narrow
|
target (`.pin-btn`, `#interval-select`, the debug-log label and, on narrow
|
||||||
viewports, `#pause-btn`). The pin button's hit area grows via matching
|
viewports, `#pause-btn`). The pin button's hit area grows via matching
|
||||||
|
|||||||
+11
-5
@@ -42,11 +42,17 @@ Internal packages in `internal/` follow standard Go project layout:
|
|||||||
|
|
||||||
### Configuration
|
### Configuration
|
||||||
|
|
||||||
| Variable | Default | Description |
|
| Variable | Default | Description |
|
||||||
| ---------- | ------------------ | --------------------------------- |
|
| ----------------- | -------------------- | -------------------------------------------------------------------------------------------------------- |
|
||||||
| `PORT` | `8080` | HTTP listen port |
|
| `PORT` | `8080` | HTTP listen port |
|
||||||
| `DATA_DIR` | `./data/reports` | Directory for compressed reports |
|
| `DATA_DIR` | `./data/reports` | Directory for compressed reports |
|
||||||
| `DEBUG` | `false` | Enable debug logging |
|
| `DEBUG` | `false` | Enable debug logging |
|
||||||
|
| `TRUSTED_PROXIES` | loopback + RFC1918 | Comma-separated CIDRs whose `X-Forwarded-For` / `X-Real-IP` headers are trusted for client IP resolution |
|
||||||
|
|
||||||
|
`TRUSTED_PROXIES` defaults to `127.0.0.1/32,::1/128,10.0.0.0/8,172.16.0.0/12,192.168.0.0/16`.
|
||||||
|
The loopback entries cover the reverse proxy that shares the container; the
|
||||||
|
RFC1918 ranges match `nginx.conf`. A request whose direct peer is outside this
|
||||||
|
set has its forwarded headers ignored, and the direct peer is logged instead.
|
||||||
|
|
||||||
### Report storage
|
### Report storage
|
||||||
|
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ package config
|
|||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
|
"strings"
|
||||||
|
|
||||||
"sneak.berlin/go/netwatch/internal/globals"
|
"sneak.berlin/go/netwatch/internal/globals"
|
||||||
"sneak.berlin/go/netwatch/internal/logger"
|
"sneak.berlin/go/netwatch/internal/logger"
|
||||||
@@ -14,6 +15,14 @@ import (
|
|||||||
"go.uber.org/fx"
|
"go.uber.org/fx"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// defaultTrustedProxies lists the networks whose forwarded
|
||||||
|
// headers are honoured by default. It covers the RFC1918
|
||||||
|
// ranges (to match nginx.conf) plus IPv4 and IPv6 loopback,
|
||||||
|
// because the reverse proxy shares the container and reaches
|
||||||
|
// the backend over loopback.
|
||||||
|
const defaultTrustedProxies = "127.0.0.1/32,::1/128," +
|
||||||
|
"10.0.0.0/8,172.16.0.0/12,192.168.0.0/16"
|
||||||
|
|
||||||
// Params defines the dependencies for Config.
|
// Params defines the dependencies for Config.
|
||||||
type Params struct {
|
type Params struct {
|
||||||
fx.In
|
fx.In
|
||||||
@@ -30,6 +39,7 @@ type Config struct {
|
|||||||
MetricsUsername string
|
MetricsUsername string
|
||||||
Port int
|
Port int
|
||||||
SentryDSN string
|
SentryDSN string
|
||||||
|
TrustedProxies []string
|
||||||
log *slog.Logger
|
log *slog.Logger
|
||||||
params *Params
|
params *Params
|
||||||
}
|
}
|
||||||
@@ -56,6 +66,7 @@ func New(
|
|||||||
viper.SetDefault("SENTRY_DSN", "")
|
viper.SetDefault("SENTRY_DSN", "")
|
||||||
viper.SetDefault("METRICS_USERNAME", "")
|
viper.SetDefault("METRICS_USERNAME", "")
|
||||||
viper.SetDefault("METRICS_PASSWORD", "")
|
viper.SetDefault("METRICS_PASSWORD", "")
|
||||||
|
viper.SetDefault("TRUSTED_PROXIES", defaultTrustedProxies)
|
||||||
|
|
||||||
err := viper.ReadInConfig()
|
err := viper.ReadInConfig()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -73,6 +84,7 @@ func New(
|
|||||||
MetricsUsername: viper.GetString("METRICS_USERNAME"),
|
MetricsUsername: viper.GetString("METRICS_USERNAME"),
|
||||||
Port: viper.GetInt("PORT"),
|
Port: viper.GetInt("PORT"),
|
||||||
SentryDSN: viper.GetString("SENTRY_DSN"),
|
SentryDSN: viper.GetString("SENTRY_DSN"),
|
||||||
|
TrustedProxies: splitList(viper.GetString("TRUSTED_PROXIES")),
|
||||||
log: log,
|
log: log,
|
||||||
params: ¶ms,
|
params: ¶ms,
|
||||||
}
|
}
|
||||||
@@ -84,3 +96,19 @@ func New(
|
|||||||
|
|
||||||
return s, nil
|
return s, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// splitList turns a comma-separated setting into a trimmed
|
||||||
|
// slice, dropping empty entries.
|
||||||
|
func splitList(raw string) []string {
|
||||||
|
parts := strings.Split(raw, ",")
|
||||||
|
|
||||||
|
out := make([]string, 0, len(parts))
|
||||||
|
for _, p := range parts {
|
||||||
|
p = strings.TrimSpace(p)
|
||||||
|
if p != "" {
|
||||||
|
out = append(out, p)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,21 @@
|
|||||||
|
package middleware
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net/http"
|
||||||
|
"net/netip"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Test-only wrappers exposing unexported helpers to the
|
||||||
|
// external middleware_test package.
|
||||||
|
|
||||||
|
func ClientIP(
|
||||||
|
remoteAddr string,
|
||||||
|
header http.Header,
|
||||||
|
trusted []netip.Prefix,
|
||||||
|
) string {
|
||||||
|
return clientIP(remoteAddr, header, trusted)
|
||||||
|
}
|
||||||
|
|
||||||
|
func ParseTrustedProxies(cidrs []string) ([]netip.Prefix, error) {
|
||||||
|
return parseTrustedProxies(cidrs)
|
||||||
|
}
|
||||||
@@ -3,9 +3,12 @@
|
|||||||
package middleware
|
package middleware
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"net/netip"
|
||||||
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"sneak.berlin/go/netwatch/internal/config"
|
"sneak.berlin/go/netwatch/internal/config"
|
||||||
@@ -19,6 +22,15 @@ import (
|
|||||||
|
|
||||||
const corsMaxAgeSec = 300
|
const corsMaxAgeSec = 300
|
||||||
|
|
||||||
|
// Security header values. The backend is a JSON API with no
|
||||||
|
// HTML surface, so the CSP forbids every resource type and
|
||||||
|
// framing outright.
|
||||||
|
const (
|
||||||
|
hstsValue = "max-age=31536000; includeSubDomains"
|
||||||
|
cspValue = "default-src 'none'; frame-ancestors 'none'"
|
||||||
|
permissionsPolicyValue = "camera=(), microphone=(), geolocation=()"
|
||||||
|
)
|
||||||
|
|
||||||
// Params defines the dependencies for Middleware.
|
// Params defines the dependencies for Middleware.
|
||||||
type Params struct {
|
type Params struct {
|
||||||
fx.In
|
fx.In
|
||||||
@@ -30,8 +42,9 @@ type Params struct {
|
|||||||
|
|
||||||
// Middleware holds shared state for middleware factories.
|
// Middleware holds shared state for middleware factories.
|
||||||
type Middleware struct {
|
type Middleware struct {
|
||||||
log *slog.Logger
|
log *slog.Logger
|
||||||
params *Params
|
params *Params
|
||||||
|
trustedProxies []netip.Prefix
|
||||||
}
|
}
|
||||||
|
|
||||||
// New creates a Middleware instance.
|
// New creates a Middleware instance.
|
||||||
@@ -39,13 +52,38 @@ func New(
|
|||||||
_ fx.Lifecycle,
|
_ fx.Lifecycle,
|
||||||
params Params,
|
params Params,
|
||||||
) (*Middleware, error) {
|
) (*Middleware, error) {
|
||||||
|
trusted, err := parseTrustedProxies(params.Config.TrustedProxies)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
s := new(Middleware)
|
s := new(Middleware)
|
||||||
s.params = ¶ms
|
s.params = ¶ms
|
||||||
s.log = params.Logger.Get()
|
s.log = params.Logger.Get()
|
||||||
|
s.trustedProxies = trusted
|
||||||
|
|
||||||
return s, nil
|
return s, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// parseTrustedProxies converts CIDR strings into prefixes,
|
||||||
|
// failing fast on any malformed entry.
|
||||||
|
func parseTrustedProxies(cidrs []string) ([]netip.Prefix, error) {
|
||||||
|
prefixes := make([]netip.Prefix, 0, len(cidrs))
|
||||||
|
|
||||||
|
for _, cidr := range cidrs {
|
||||||
|
prefix, err := netip.ParsePrefix(cidr)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf(
|
||||||
|
"trusted proxy %q: %w", cidr, err,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
prefixes = append(prefixes, prefix.Masked())
|
||||||
|
}
|
||||||
|
|
||||||
|
return prefixes, nil
|
||||||
|
}
|
||||||
|
|
||||||
type loggingResponseWriter struct {
|
type loggingResponseWriter struct {
|
||||||
http.ResponseWriter
|
http.ResponseWriter
|
||||||
|
|
||||||
@@ -72,6 +110,70 @@ func ipFromHostPort(hostPort string) string {
|
|||||||
return host
|
return host
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// clientIP resolves the caller's address. X-Forwarded-For and
|
||||||
|
// X-Real-IP are honoured only when the direct peer is a
|
||||||
|
// trusted proxy; otherwise the direct peer is returned so a
|
||||||
|
// spoofed header cannot forge the logged address.
|
||||||
|
func clientIP(
|
||||||
|
remoteAddr string,
|
||||||
|
header http.Header,
|
||||||
|
trusted []netip.Prefix,
|
||||||
|
) string {
|
||||||
|
peer := ipFromHostPort(remoteAddr)
|
||||||
|
|
||||||
|
if !addrInAny(peer, trusted) {
|
||||||
|
return peer
|
||||||
|
}
|
||||||
|
|
||||||
|
if xff := firstForwardedFor(header.Get("X-Forwarded-For")); xff != "" {
|
||||||
|
return xff
|
||||||
|
}
|
||||||
|
|
||||||
|
if xr := strings.TrimSpace(header.Get("X-Real-IP")); validIP(xr) {
|
||||||
|
return xr
|
||||||
|
}
|
||||||
|
|
||||||
|
return peer
|
||||||
|
}
|
||||||
|
|
||||||
|
// firstForwardedFor returns the left-most valid address in an
|
||||||
|
// X-Forwarded-For list (the original client), or "" if none.
|
||||||
|
func firstForwardedFor(value string) string {
|
||||||
|
for part := range strings.SplitSeq(value, ",") {
|
||||||
|
candidate := strings.TrimSpace(part)
|
||||||
|
if validIP(candidate) {
|
||||||
|
return candidate
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
|
||||||
|
func validIP(s string) bool {
|
||||||
|
_, err := netip.ParseAddr(s)
|
||||||
|
|
||||||
|
return err == nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// addrInAny reports whether s parses as an address contained
|
||||||
|
// in any of the trusted prefixes.
|
||||||
|
func addrInAny(s string, trusted []netip.Prefix) bool {
|
||||||
|
addr, err := netip.ParseAddr(s)
|
||||||
|
if err != nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
addr = addr.Unmap()
|
||||||
|
|
||||||
|
for _, prefix := range trusted {
|
||||||
|
if prefix.Contains(addr) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
// Logging returns middleware that logs each request with
|
// Logging returns middleware that logs each request with
|
||||||
// timing, status code, and client information.
|
// timing, status code, and client information.
|
||||||
func (s *Middleware) Logging() func(http.Handler) http.Handler {
|
func (s *Middleware) Logging() func(http.Handler) http.Handler {
|
||||||
@@ -96,7 +198,11 @@ func (s *Middleware) Logging() func(http.Handler) http.Handler {
|
|||||||
"referer", r.Referer(),
|
"referer", r.Referer(),
|
||||||
"proto", r.Proto,
|
"proto", r.Proto,
|
||||||
"remote_ip",
|
"remote_ip",
|
||||||
ipFromHostPort(r.RemoteAddr),
|
clientIP(
|
||||||
|
r.RemoteAddr,
|
||||||
|
r.Header,
|
||||||
|
s.trustedProxies,
|
||||||
|
),
|
||||||
"status", lrw.statusCode,
|
"status", lrw.statusCode,
|
||||||
"latency_ms",
|
"latency_ms",
|
||||||
latency.Milliseconds(),
|
latency.Milliseconds(),
|
||||||
@@ -109,6 +215,27 @@ func (s *Middleware) Logging() func(http.Handler) http.Handler {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SecurityHeaders returns middleware that sets response
|
||||||
|
// security headers. It runs before CORS so the headers are
|
||||||
|
// present on preflight responses the CORS handler writes.
|
||||||
|
func (s *Middleware) SecurityHeaders() func(http.Handler) http.Handler {
|
||||||
|
return func(next http.Handler) http.Handler {
|
||||||
|
return http.HandlerFunc(
|
||||||
|
func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
h := w.Header()
|
||||||
|
h.Set("Strict-Transport-Security", hstsValue)
|
||||||
|
h.Set("Content-Security-Policy", cspValue)
|
||||||
|
h.Set("X-Frame-Options", "DENY")
|
||||||
|
h.Set("X-Content-Type-Options", "nosniff")
|
||||||
|
h.Set("Referrer-Policy", "no-referrer")
|
||||||
|
h.Set("Permissions-Policy", permissionsPolicyValue)
|
||||||
|
|
||||||
|
next.ServeHTTP(w, r)
|
||||||
|
},
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// CORS returns middleware that adds permissive CORS headers.
|
// CORS returns middleware that adds permissive CORS headers.
|
||||||
func (s *Middleware) CORS() func(http.Handler) http.Handler {
|
func (s *Middleware) CORS() func(http.Handler) http.Handler {
|
||||||
return cors.Handler(cors.Options{
|
return cors.Handler(cors.Options{
|
||||||
|
|||||||
@@ -0,0 +1,139 @@
|
|||||||
|
package middleware_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"net/netip"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"sneak.berlin/go/netwatch/internal/middleware"
|
||||||
|
)
|
||||||
|
|
||||||
|
func mustPrefixes(t *testing.T, cidrs ...string) []netip.Prefix {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
prefixes, err := middleware.ParseTrustedProxies(cidrs)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ParseTrustedProxies(%v): %v", cidrs, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return prefixes
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestParseTrustedProxiesRejectsMalformed(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
_, err := middleware.ParseTrustedProxies([]string{"not-a-cidr"})
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("expected error for malformed CIDR, got nil")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type clientIPCase struct {
|
||||||
|
name string
|
||||||
|
remoteAddr string
|
||||||
|
xff string
|
||||||
|
xRealIP string
|
||||||
|
want string
|
||||||
|
}
|
||||||
|
|
||||||
|
func clientIPCases() []clientIPCase {
|
||||||
|
return []clientIPCase{
|
||||||
|
{
|
||||||
|
name: "trusted proxy uses forwarded-for",
|
||||||
|
remoteAddr: "127.0.0.1:5000",
|
||||||
|
xff: "203.0.113.7",
|
||||||
|
want: "203.0.113.7",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "trusted proxy uses left-most of chain",
|
||||||
|
remoteAddr: "10.1.2.3:5000",
|
||||||
|
xff: "203.0.113.7, 10.1.2.3",
|
||||||
|
want: "203.0.113.7",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "trusted proxy falls back to x-real-ip",
|
||||||
|
remoteAddr: "127.0.0.1:5000",
|
||||||
|
xRealIP: "203.0.113.9",
|
||||||
|
want: "203.0.113.9",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "untrusted peer ignores forwarded-for",
|
||||||
|
remoteAddr: "198.51.100.4:5000",
|
||||||
|
xff: "203.0.113.7",
|
||||||
|
want: "198.51.100.4",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "untrusted peer ignores x-real-ip",
|
||||||
|
remoteAddr: "198.51.100.4:5000",
|
||||||
|
xRealIP: "203.0.113.9",
|
||||||
|
want: "198.51.100.4",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "trusted proxy with no headers uses peer",
|
||||||
|
remoteAddr: "10.1.2.3:5000",
|
||||||
|
want: "10.1.2.3",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "trusted proxy with garbage header uses peer",
|
||||||
|
remoteAddr: "127.0.0.1:5000",
|
||||||
|
xff: "not-an-ip",
|
||||||
|
want: "127.0.0.1",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestClientIP(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
trusted := mustPrefixes(t, "127.0.0.1/32", "::1/128", "10.0.0.0/8")
|
||||||
|
|
||||||
|
for _, tc := range clientIPCases() {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
header := http.Header{}
|
||||||
|
if tc.xff != "" {
|
||||||
|
header.Set("X-Forwarded-For", tc.xff)
|
||||||
|
}
|
||||||
|
|
||||||
|
if tc.xRealIP != "" {
|
||||||
|
header.Set("X-Real-IP", tc.xRealIP)
|
||||||
|
}
|
||||||
|
|
||||||
|
got := middleware.ClientIP(tc.remoteAddr, header, trusted)
|
||||||
|
if got != tc.want {
|
||||||
|
t.Errorf("ClientIP() = %q, want %q", got, tc.want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSecurityHeaders(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
handler := (&middleware.Middleware{}).SecurityHeaders()(
|
||||||
|
http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
req := httptest.NewRequest(http.MethodGet, "/", http.NoBody)
|
||||||
|
handler.ServeHTTP(rec, req)
|
||||||
|
|
||||||
|
want := map[string]string{
|
||||||
|
"Strict-Transport-Security": "max-age=31536000; includeSubDomains",
|
||||||
|
"Content-Security-Policy": "default-src 'none'; frame-ancestors 'none'",
|
||||||
|
"X-Frame-Options": "DENY",
|
||||||
|
"X-Content-Type-Options": "nosniff",
|
||||||
|
"Referrer-Policy": "no-referrer",
|
||||||
|
"Permissions-Policy": "camera=(), microphone=(), geolocation=()",
|
||||||
|
}
|
||||||
|
|
||||||
|
for name, value := range want {
|
||||||
|
if got := rec.Header().Get(name); got != value {
|
||||||
|
t.Errorf("header %s = %q, want %q", name, got, value)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -40,11 +40,12 @@ type Params struct {
|
|||||||
// Buffer accumulates JSON lines in memory and flushes them
|
// Buffer accumulates JSON lines in memory and flushes them
|
||||||
// to zstd-compressed files on disk.
|
// to zstd-compressed files on disk.
|
||||||
type Buffer struct {
|
type Buffer struct {
|
||||||
buf bytes.Buffer
|
buf bytes.Buffer
|
||||||
dataDir string
|
dataDir string
|
||||||
done chan struct{}
|
done chan struct{}
|
||||||
log *slog.Logger
|
log *slog.Logger
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
|
stopOnce sync.Once
|
||||||
}
|
}
|
||||||
|
|
||||||
// New creates a Buffer and registers lifecycle hooks to
|
// New creates a Buffer and registers lifecycle hooks to
|
||||||
@@ -76,8 +77,13 @@ func New(
|
|||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
OnStop: func(_ context.Context) error {
|
OnStop: func(_ context.Context) error {
|
||||||
close(b.done)
|
// stopOnce makes OnStop idempotent: a second
|
||||||
b.flushLocked()
|
// invocation must not close an already-closed channel
|
||||||
|
// (which would panic) or flush again.
|
||||||
|
b.stopOnce.Do(func() {
|
||||||
|
close(b.done)
|
||||||
|
b.flushLocked()
|
||||||
|
})
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -1,13 +1,77 @@
|
|||||||
package reportbuf_test
|
package reportbuf_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"os"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
_ "sneak.berlin/go/netwatch/internal/reportbuf"
|
"sneak.berlin/go/netwatch/internal/config"
|
||||||
|
"sneak.berlin/go/netwatch/internal/globals"
|
||||||
|
"sneak.berlin/go/netwatch/internal/logger"
|
||||||
|
"sneak.berlin/go/netwatch/internal/reportbuf"
|
||||||
|
|
||||||
|
"go.uber.org/fx"
|
||||||
|
"go.uber.org/fx/fxtest"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestImport(t *testing.T) {
|
// TestFlushOnShutdown proves the flush-on-shutdown path: a
|
||||||
t.Parallel()
|
// report appended after start but before the periodic flush
|
||||||
// Compilation check — verifies the package parses
|
// window must reach disk when the fx lifecycle stops. This is
|
||||||
// and all imports resolve.
|
// the exact case that silent data loss on restart used to
|
||||||
|
// destroy.
|
||||||
|
func TestFlushOnShutdown(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
t.Setenv("DATA_DIR", dir)
|
||||||
|
|
||||||
|
var buf *reportbuf.Buffer
|
||||||
|
|
||||||
|
app := fxtest.New(t,
|
||||||
|
fx.Provide(
|
||||||
|
globals.New,
|
||||||
|
logger.New,
|
||||||
|
config.New,
|
||||||
|
reportbuf.New,
|
||||||
|
),
|
||||||
|
fx.Populate(&buf),
|
||||||
|
)
|
||||||
|
|
||||||
|
app.RequireStart()
|
||||||
|
|
||||||
|
err := buf.Append(map[string]string{"probe": "shutdown"})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("append report: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// RequireStop runs the reportbuf OnStop hook, which is the
|
||||||
|
// only code path that flushes buffered reports on shutdown.
|
||||||
|
app.RequireStop()
|
||||||
|
|
||||||
|
if !hasReportFile(t, dir) {
|
||||||
|
t.Fatal("no report file on disk after shutdown; " +
|
||||||
|
"the buffered report was lost")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func hasReportFile(t *testing.T, dir string) bool {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
entries, err := os.ReadDir(dir)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("read data dir: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, e := range entries {
|
||||||
|
if strings.HasSuffix(e.Name(), ".jsonl.zst") {
|
||||||
|
info, statErr := e.Info()
|
||||||
|
if statErr != nil {
|
||||||
|
t.Fatalf("stat %s: %v", e.Name(), statErr)
|
||||||
|
}
|
||||||
|
|
||||||
|
if info.Size() > 0 {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return false
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,39 +5,62 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"go.uber.org/fx"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
readTimeout = 10 * time.Second
|
readTimeout = 10 * time.Second
|
||||||
writeTimeout = 10 * time.Second
|
readHeaderTimeout = 5 * time.Second
|
||||||
maxHeaderBytes = 1 << 20 // 1 MiB
|
idleTimeout = 60 * time.Second
|
||||||
|
maxHeaderBytes = 1 << 20 // 1 MiB
|
||||||
|
|
||||||
|
// requestTimeout (routes.go) is the single per-request
|
||||||
|
// processing budget, enforced by chi's middleware.Timeout.
|
||||||
|
// writeTimeout must exceed that budget so a handler can write
|
||||||
|
// its 503 when the chi timeout fires; if it were shorter the
|
||||||
|
// server would abort the write first and the chi budget would
|
||||||
|
// be unreachable dead configuration.
|
||||||
|
writeTimeout = requestTimeout + 5*time.Second
|
||||||
)
|
)
|
||||||
|
|
||||||
func (s *Server) serveUntilShutdown() {
|
// newHTTPServer constructs the http.Server. It performs no I/O
|
||||||
|
// and does not start listening.
|
||||||
|
func (s *Server) newHTTPServer() *http.Server {
|
||||||
listenAddr := fmt.Sprintf(":%d", s.params.Config.Port)
|
listenAddr := fmt.Sprintf(":%d", s.params.Config.Port)
|
||||||
|
|
||||||
s.httpServer = &http.Server{
|
return &http.Server{
|
||||||
Addr: listenAddr,
|
Addr: listenAddr,
|
||||||
Handler: s,
|
Handler: s,
|
||||||
MaxHeaderBytes: maxHeaderBytes,
|
MaxHeaderBytes: maxHeaderBytes,
|
||||||
ReadTimeout: readTimeout,
|
ReadTimeout: readTimeout,
|
||||||
WriteTimeout: writeTimeout,
|
ReadHeaderTimeout: readHeaderTimeout,
|
||||||
|
WriteTimeout: writeTimeout,
|
||||||
|
IdleTimeout: idleTimeout,
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
s.SetupRoutes()
|
// listenAndServe runs the listener until the server is shut
|
||||||
|
// down. A genuine listen failure (not the expected
|
||||||
|
// ErrServerClosed from a clean shutdown) requests process
|
||||||
|
// shutdown through fx with a non-zero exit code, so the failure
|
||||||
|
// is visible to any supervisor.
|
||||||
|
func (s *Server) listenAndServe() {
|
||||||
s.log.Info("http begin listen",
|
s.log.Info("http begin listen",
|
||||||
"listenaddr", listenAddr,
|
"listenaddr", s.httpServer.Addr,
|
||||||
"version", s.params.Globals.Version,
|
"version", s.params.Globals.Version,
|
||||||
"buildarch", s.params.Globals.Buildarch,
|
"buildarch", s.params.Globals.Buildarch,
|
||||||
)
|
)
|
||||||
|
|
||||||
err := s.httpServer.ListenAndServe()
|
err := s.httpServer.ListenAndServe()
|
||||||
if err != nil && !errors.Is(err, http.ErrServerClosed) {
|
if err == nil || errors.Is(err, http.ErrServerClosed) {
|
||||||
s.log.Error("listen error", "error", err)
|
return
|
||||||
|
}
|
||||||
|
|
||||||
if s.cancelFunc != nil {
|
s.log.Error("listen error", "error", err)
|
||||||
s.cancelFunc()
|
|
||||||
}
|
shutdownErr := s.shutdowner.Shutdown(fx.ExitCode(1))
|
||||||
|
if shutdownErr != nil {
|
||||||
|
s.log.Error("request shutdown failed", "error", shutdownErr)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ func (s *Server) SetupRoutes() {
|
|||||||
s.router.Use(middleware.Recoverer)
|
s.router.Use(middleware.Recoverer)
|
||||||
s.router.Use(middleware.RequestID)
|
s.router.Use(middleware.RequestID)
|
||||||
s.router.Use(s.mw.Logging())
|
s.router.Use(s.mw.Logging())
|
||||||
|
s.router.Use(s.mw.SecurityHeaders())
|
||||||
s.router.Use(s.mw.CORS())
|
s.router.Use(s.mw.CORS())
|
||||||
s.router.Use(middleware.Timeout(requestTimeout))
|
s.router.Use(middleware.Timeout(requestTimeout))
|
||||||
|
|
||||||
|
|||||||
@@ -1,16 +1,14 @@
|
|||||||
// Package server provides the HTTP server lifecycle,
|
// Package server provides the HTTP server lifecycle,
|
||||||
// including startup, routing, signal handling, and graceful
|
// including startup, routing, and graceful shutdown. The
|
||||||
// shutdown.
|
// process lifetime is owned by fx: shutdown is requested
|
||||||
|
// through fx.Shutdowner so every component's OnStop hook runs
|
||||||
|
// in dependency order.
|
||||||
package server
|
package server
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"net/http"
|
"net/http"
|
||||||
"os"
|
|
||||||
"os/signal"
|
|
||||||
"syscall"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"sneak.berlin/go/netwatch/internal/config"
|
"sneak.berlin/go/netwatch/internal/config"
|
||||||
"sneak.berlin/go/netwatch/internal/globals"
|
"sneak.berlin/go/netwatch/internal/globals"
|
||||||
@@ -31,19 +29,18 @@ type Params struct {
|
|||||||
Handlers *handlers.Handlers
|
Handlers *handlers.Handlers
|
||||||
Logger *logger.Logger
|
Logger *logger.Logger
|
||||||
Middleware *middleware.Middleware
|
Middleware *middleware.Middleware
|
||||||
|
Shutdowner fx.Shutdowner
|
||||||
}
|
}
|
||||||
|
|
||||||
// Server is the top-level HTTP server orchestrator.
|
// Server is the top-level HTTP server orchestrator.
|
||||||
type Server struct {
|
type Server struct {
|
||||||
cancelFunc context.CancelFunc
|
h *handlers.Handlers
|
||||||
exitCode int
|
httpServer *http.Server
|
||||||
h *handlers.Handlers
|
log *slog.Logger
|
||||||
httpServer *http.Server
|
mw *middleware.Middleware
|
||||||
log *slog.Logger
|
params Params
|
||||||
mw *middleware.Middleware
|
router *chi.Mux
|
||||||
params Params
|
shutdowner fx.Shutdowner
|
||||||
router *chi.Mux
|
|
||||||
startupTime time.Time
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// New creates a Server and registers lifecycle hooks for
|
// New creates a Server and registers lifecycle hooks for
|
||||||
@@ -57,23 +54,25 @@ func New(
|
|||||||
s.mw = params.Middleware
|
s.mw = params.Middleware
|
||||||
s.h = params.Handlers
|
s.h = params.Handlers
|
||||||
s.log = params.Logger.Get()
|
s.log = params.Logger.Get()
|
||||||
|
s.shutdowner = params.Shutdowner
|
||||||
|
|
||||||
lc.Append(fx.Hook{
|
lc.Append(fx.Hook{
|
||||||
OnStart: func(_ context.Context) error {
|
OnStart: func(_ context.Context) error {
|
||||||
s.startupTime = time.Now().UTC()
|
// Build the router and http.Server synchronously
|
||||||
|
// here, before spawning the serving goroutine, so
|
||||||
|
// httpServer is fully constructed by the time OnStop
|
||||||
|
// (or an early signal) can read it. fx guarantees
|
||||||
|
// OnStart returns before OnStop runs, so no
|
||||||
|
// synchronization or nil check is needed at shutdown.
|
||||||
|
s.SetupRoutes()
|
||||||
|
s.httpServer = s.newHTTPServer()
|
||||||
|
|
||||||
go func() { //nolint:contextcheck // fx OnStart ctx is startup-only; run() creates its own
|
go s.listenAndServe()
|
||||||
s.run()
|
|
||||||
}()
|
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
OnStop: func(_ context.Context) error {
|
OnStop: func(ctx context.Context) error {
|
||||||
if s.cancelFunc != nil {
|
return s.shutdown(ctx)
|
||||||
s.cancelFunc()
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -88,60 +87,17 @@ func (s *Server) ServeHTTP(
|
|||||||
s.router.ServeHTTP(w, r)
|
s.router.ServeHTTP(w, r)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Server) run() {
|
// shutdown gracefully stops the HTTP server within the
|
||||||
exitCode := s.serve()
|
// deadline of the context fx provides for OnStop.
|
||||||
os.Exit(exitCode)
|
func (s *Server) shutdown(ctx context.Context) error {
|
||||||
}
|
err := s.httpServer.Shutdown(ctx)
|
||||||
|
|
||||||
func (s *Server) serve() int {
|
|
||||||
var ctx context.Context //nolint:wsl // ctx must be declared before multi-assign
|
|
||||||
|
|
||||||
ctx, s.cancelFunc = context.WithCancel(
|
|
||||||
context.Background(),
|
|
||||||
)
|
|
||||||
|
|
||||||
go func() {
|
|
||||||
c := make(chan os.Signal, 1)
|
|
||||||
|
|
||||||
signal.Ignore(syscall.SIGPIPE)
|
|
||||||
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
|
|
||||||
|
|
||||||
sig := <-c
|
|
||||||
s.log.Info("signal received", "signal", sig)
|
|
||||||
|
|
||||||
if s.cancelFunc != nil {
|
|
||||||
s.cancelFunc()
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
go func() {
|
|
||||||
s.serveUntilShutdown()
|
|
||||||
}()
|
|
||||||
|
|
||||||
<-ctx.Done()
|
|
||||||
s.cleanShutdown()
|
|
||||||
|
|
||||||
return s.exitCode
|
|
||||||
}
|
|
||||||
|
|
||||||
const shutdownTimeout = 5 * time.Second
|
|
||||||
|
|
||||||
func (s *Server) cleanShutdown() {
|
|
||||||
s.exitCode = 0
|
|
||||||
|
|
||||||
ctxShutdown, shutdownCancel := context.WithTimeout(
|
|
||||||
context.Background(),
|
|
||||||
shutdownTimeout,
|
|
||||||
)
|
|
||||||
defer shutdownCancel()
|
|
||||||
|
|
||||||
err := s.httpServer.Shutdown(ctxShutdown)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.log.Error(
|
s.log.Error("server clean shutdown failed", "error", err)
|
||||||
"server clean shutdown failed",
|
|
||||||
"error", err,
|
return err
|
||||||
)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
s.log.Info("server stopped")
|
s.log.Info("server stopped")
|
||||||
|
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user