Compare commits
1
Commits
next
..
dfca772035
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dfca772035 |
+2
-12
@@ -73,8 +73,7 @@
|
||||
|
||||
# This repository's host-built artifacts: `make build` writes bin/, `go
|
||||
# build` in the root the binaries, `make run` log.txt, `make asupdate` its
|
||||
# .tmp file; logs, test binaries, coverage profiles, source archives, and
|
||||
# the binaries and libraries a build for another platform writes.
|
||||
# .tmp file; test binaries, coverage profiles and source archives.
|
||||
/bin
|
||||
/routewatch
|
||||
/asinfo-gen
|
||||
@@ -83,24 +82,15 @@
|
||||
/out
|
||||
/pkg/asinfo/asdata.json
|
||||
**/*.tar.zst
|
||||
**/*.log
|
||||
**/*.test
|
||||
**/*.out
|
||||
**/*.tmp
|
||||
**/*.exe
|
||||
**/*.dll
|
||||
**/*.so
|
||||
**/*.dylib
|
||||
|
||||
# Local databases. The image carries a source archive of the whole build
|
||||
# context, so these would otherwise ship inside it. The daemon's database
|
||||
# is db.sqlite, with its -wal and -shm files while it runs.
|
||||
# context, so these would otherwise ship inside it.
|
||||
**/*.db
|
||||
**/*.db-journal
|
||||
**/*.db-wal
|
||||
**/*.sqlite
|
||||
**/*.sqlite-wal
|
||||
**/*.sqlite-shm
|
||||
|
||||
# A local Go workspace points at directories outside the build context.
|
||||
/go.work
|
||||
|
||||
+1
-5
@@ -67,14 +67,10 @@ node_modules/
|
||||
go.work
|
||||
go.work.sum
|
||||
|
||||
# Local databases, temporary files and debug output. The daemon's
|
||||
# database is db.sqlite, with its -wal and -shm files while it runs.
|
||||
# Local databases, temporary files and debug output.
|
||||
*.db
|
||||
*.db-journal
|
||||
*.db-wal
|
||||
*.sqlite
|
||||
*.sqlite-wal
|
||||
*.sqlite-shm
|
||||
*.tmp
|
||||
log.txt
|
||||
out
|
||||
|
||||
@@ -5,22 +5,6 @@ table in real time: it streams BGP UPDATE messages from the RIPE RIS Live
|
||||
service, maintains a live routing table in SQLite, and provides HTTP APIs for
|
||||
querying routing information.
|
||||
|
||||
## Rationale
|
||||
|
||||
RIPE RIS Live streams the BGP updates that RIPE's route collectors receive, as
|
||||
they happen, from the moment a client connects. It reports changes, not the
|
||||
routing table they add up to, so it cannot say which AS an IP address is routed
|
||||
to now or which routes a prefix has. RouteWatch applies the feed's updates to
|
||||
live routing tables in its own SQLite database and answers IP-to-AS lookups and
|
||||
prefix pages from them.
|
||||
|
||||
The feed sends no table when a client connects, so these tables hold only the
|
||||
routes the feed has announced since RouteWatch first started: an IP-to-AS lookup
|
||||
finds nothing for an address whose routes have not been announced since then.
|
||||
The tables also miss the updates RouteWatch drops when it falls behind under
|
||||
load and those sent while it is stopped, so a lookup or a prefix page can show a
|
||||
route the feed has withdrawn.
|
||||
|
||||
## Features
|
||||
|
||||
- Real-time streaming of BGP updates from RIPE RIS Live
|
||||
@@ -52,8 +36,6 @@ DEBUG=routewatch ./routewatch
|
||||
|
||||
## HTTP Endpoints
|
||||
|
||||
Every route answers `HEAD` as it answers `GET`, without the body.
|
||||
|
||||
### Web Interface
|
||||
|
||||
- `GET /` - Home page (HTML): headline figures, and lookups by AS number and by
|
||||
@@ -108,11 +90,8 @@ and 4 seconds for the database reads behind `/status.json` and `/api/v1/stats`.
|
||||
An AS number, prefix or mask length that does not parse answers 400, as does a
|
||||
mask length out of range for its IP version. An AS the database does not hold,
|
||||
or a prefix with no live route, answers 404. The HTML pages give these errors as
|
||||
plain text. Under `/api/v1` and `/ip/`, a path no route matches answers 404 with
|
||||
the message `Not found`, and `POST`, `PUT`, `PATCH`, `DELETE`, `OPTIONS`,
|
||||
`TRACE` or `CONNECT` answers 405 with `Method not allowed`. A 500 answer says
|
||||
only `Internal server error`, and a host name that does not resolve only
|
||||
`could not resolve host name`; the detail goes to the log.
|
||||
plain text. A 500 answer says only `Internal server error`, and a host name that
|
||||
does not resolve only `could not resolve host name`; the detail goes to the log.
|
||||
|
||||
Every answer carries `Strict-Transport-Security`, `Content-Security-Policy`,
|
||||
`X-Frame-Options`, `X-Content-Type-Options`, `Referrer-Policy` and
|
||||
@@ -211,11 +190,11 @@ routewatch/
|
||||
- **ASHandler**: Tracks all ASNs seen in AS paths
|
||||
- **PeerHandler**: Records BGP peer information
|
||||
- **PrefixHandler**: Maintains live routing table (upserts on announcement,
|
||||
deletes on withdrawal or on an announcement whose AS path is empty or
|
||||
holds AS 0 anywhere, deletes all of a peer's routes when its BGP session
|
||||
goes down). Routes are kept per peer IP, so when one peer IP has sessions
|
||||
on two collectors, a session going down on one also removes the routes the
|
||||
other still carries, until they are announced again.
|
||||
deletes on withdrawal or on an announcement whose AS path is empty or ends
|
||||
in AS 0, deletes all of a peer's routes when its BGP session goes down).
|
||||
Routes are kept per peer IP, so when one peer IP has sessions on two
|
||||
collectors, a session going down on one also removes the routes the other
|
||||
still carries, until they are announced again.
|
||||
- **PeeringHandler**: Extracts AS peering relationships from AS paths
|
||||
|
||||
5. **HTTP Request Flow**
|
||||
|
||||
@@ -169,20 +169,6 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
|
||||
return nil, fmt.Errorf("failed to create database directory: %w", err)
|
||||
}
|
||||
|
||||
// The size of the WAL file the previous run left, which Initialize logs
|
||||
// with the start-up checkpoint. It is read before the database is opened,
|
||||
// because opening it writes a frame to the WAL. There is no WAL file when
|
||||
// the database is new or the last run closed the database, which deletes
|
||||
// it.
|
||||
var walBytes int64
|
||||
|
||||
wal, err := os.Stat(dbPath + walFileSuffix)
|
||||
if err == nil {
|
||||
walBytes = wal.Size()
|
||||
} else if !errors.Is(err, os.ErrNotExist) {
|
||||
logger.Warn("Failed to read the WAL file's size on startup", "error", err)
|
||||
}
|
||||
|
||||
// Per-connection SQLite settings go in the DSN so every pooled connection
|
||||
// gets them, not just the one that runs the Initialize pragmas. _txlock=
|
||||
// immediate makes every transaction take the write lock at BEGIN. Without it
|
||||
@@ -243,7 +229,7 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
|
||||
counts: &liveCounts{},
|
||||
}
|
||||
|
||||
err = database.Initialize(walBytes)
|
||||
err = database.Initialize()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to initialize database: %w", err)
|
||||
}
|
||||
@@ -259,16 +245,8 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
|
||||
return database, nil
|
||||
}
|
||||
|
||||
// The messages Initialize logs the start-up WAL checkpoint with.
|
||||
const (
|
||||
startupCheckpointMessage = "WAL checkpoint on startup"
|
||||
startupCheckpointIncompleteMessage = "WAL checkpoint on startup incomplete"
|
||||
)
|
||||
|
||||
// Initialize creates the database schema if it doesn't exist. walBytes is the
|
||||
// size of the WAL file the previous run left, which New reads before it opens
|
||||
// the database.
|
||||
func (d *Database) Initialize(walBytes int64) error {
|
||||
// Initialize creates the database schema if it doesn't exist.
|
||||
func (d *Database) Initialize() error {
|
||||
// Set SQLite pragmas for performance. Per-connection settings (cache_size,
|
||||
// synchronous, busy_timeout, journal_mode) live in the DSN, and New sets the
|
||||
// writing connection's cache_size; temp_store is left at its default so
|
||||
@@ -288,33 +266,19 @@ func (d *Database) Initialize(walBytes int64) error {
|
||||
}
|
||||
}
|
||||
|
||||
// The start-up checkpoint copies the WAL a previous run left into the
|
||||
// database file and empties the WAL file; its frame counts then read 0,
|
||||
// so walBytes is logged with them.
|
||||
var busy, walFrames, checkpointedFrames int
|
||||
// Run WAL checkpoint on startup to consolidate any existing WAL
|
||||
var walPages, checkpointedPages, movedPages int
|
||||
|
||||
err := d.writer.QueryRowContext(context.Background(),
|
||||
"PRAGMA wal_checkpoint(TRUNCATE)",
|
||||
).Scan(&busy, &walFrames, &checkpointedFrames)
|
||||
|
||||
switch {
|
||||
case err != nil:
|
||||
).Scan(&walPages, &checkpointedPages, &movedPages)
|
||||
if err != nil {
|
||||
d.logger.Warn("Failed to checkpoint WAL on startup", "error", err)
|
||||
case busy != 0:
|
||||
// Another process using the database kept the checkpoint from
|
||||
// copying or emptying all of the WAL.
|
||||
d.logger.Warn(startupCheckpointIncompleteMessage,
|
||||
"wal_bytes", walBytes,
|
||||
"busy", busy,
|
||||
"wal_frames", walFrames,
|
||||
"checkpointed_frames", checkpointedFrames,
|
||||
)
|
||||
default:
|
||||
d.logger.Info(startupCheckpointMessage,
|
||||
"wal_bytes", walBytes,
|
||||
"busy", busy,
|
||||
"wal_frames", walFrames,
|
||||
"checkpointed_frames", checkpointedFrames,
|
||||
} else {
|
||||
d.logger.Info("WAL checkpoint on startup",
|
||||
"wal_pages", walPages,
|
||||
"checkpointed", checkpointedPages,
|
||||
"moved", movedPages,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -891,7 +855,7 @@ func insertPrefixesInTable(
|
||||
|
||||
// UpdatePrefixesBatch stores, in a single transaction, the prefixes that are
|
||||
// not stored yet, each first seen at its time in prefixes: the IPv4 prefixes
|
||||
// first and then the IPv6 prefixes. A prefix that does not parse is not stored.
|
||||
// first and then the IPv6 prefixes.
|
||||
func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
|
||||
if len(prefixes) == 0 {
|
||||
return nil
|
||||
@@ -910,11 +874,6 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
|
||||
prefixesV6 := make(map[string]time.Time)
|
||||
|
||||
for prefix, timestamp := range prefixes {
|
||||
_, err := netip.ParsePrefix(prefix)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
if detectIPVersion(prefix) == ipVersionV4 {
|
||||
prefixesV4[prefix] = timestamp
|
||||
} else {
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
@@ -1054,176 +1052,6 @@ func walFileSize(t *testing.T, db *Database) int64 {
|
||||
return wal.Size()
|
||||
}
|
||||
|
||||
// TestStartupCheckpointLogsItsFigures opens a new database, writes a route,
|
||||
// and opens the same file again while the first stays open, so that its WAL
|
||||
// file stays. The first open finds no WAL file: it logs a size of 0 and no
|
||||
// warning. The second logs the size of the WAL file the first left, and the
|
||||
// checkpoint's busy flag and frame counts, each under its own name; the
|
||||
// counts read 0 once the checkpoint has emptied the file.
|
||||
// Initialize logged the busy flag as wal_pages and the counts under each
|
||||
// other's names (https://git.eeqj.de/sneak/routewatch/issues/117).
|
||||
func TestStartupCheckpointLogsItsFigures(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cfg := &config.Config{StateDir: t.TempDir()}
|
||||
|
||||
db, lines := openLogged(t, cfg)
|
||||
|
||||
for _, line := range lines {
|
||||
if line[slog.LevelKey] == slog.LevelWarn.String() {
|
||||
t.Errorf("opening a new database logged a warning: %v", line)
|
||||
}
|
||||
}
|
||||
|
||||
line := findLine(t, lines, startupCheckpointMessage)
|
||||
if line["wal_bytes"] != 0.0 {
|
||||
t.Errorf("new database: start-up checkpoint logged wal_bytes = %v, "+
|
||||
"want 0", line["wal_bytes"])
|
||||
}
|
||||
|
||||
err := db.UpsertLiveRouteBatch([]*LiveRoute{
|
||||
mkV4Route(t, "192.0.2.0/24", lockTestASN, time.Now().UTC()),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to store a route: %v", err)
|
||||
}
|
||||
|
||||
walBytes := walFileSize(t, db)
|
||||
if walBytes == 0 {
|
||||
t.Fatal("WAL file empty after the route write, want frames in it")
|
||||
}
|
||||
|
||||
_, lines = openLogged(t, cfg)
|
||||
line = findLine(t, lines, startupCheckpointMessage)
|
||||
|
||||
for _, key := range []string{"busy", "wal_frames", "checkpointed_frames"} {
|
||||
if line[key] != 0.0 {
|
||||
t.Errorf("start-up checkpoint logged %s = %v, want 0", key, line[key])
|
||||
}
|
||||
}
|
||||
|
||||
if line["wal_bytes"] != float64(walBytes) {
|
||||
t.Errorf("start-up checkpoint logged wal_bytes = %v, want the %d "+
|
||||
"bytes the first database left", line["wal_bytes"], walBytes)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStartupCheckpointWarnsWhenIncomplete keeps a read open on a database
|
||||
// while a route is written after the read began, then opens the same file
|
||||
// again. The second open's start-up checkpoint cannot copy the frames written
|
||||
// after the read began, and gives up once the busy timeout has passed. It
|
||||
// checks that the checkpoint is logged as a warning with busy 1 and fewer
|
||||
// frames copied than the WAL holds.
|
||||
func TestStartupCheckpointWarnsWhenIncomplete(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cfg := &config.Config{StateDir: t.TempDir()}
|
||||
db, _ := openLogged(t, cfg)
|
||||
ctx := context.Background()
|
||||
|
||||
reader, err := db.db.Conn(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to open the reading connection: %v", err)
|
||||
}
|
||||
defer func() { _ = reader.Close() }()
|
||||
|
||||
// A plain BEGIN starts a transaction that only reads, unlike the BEGIN
|
||||
// IMMEDIATE of _txlock. It keeps the snapshot its first read takes until
|
||||
// it ends.
|
||||
_, err = reader.ExecContext(ctx, "BEGIN")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to begin the read transaction: %v", err)
|
||||
}
|
||||
defer func() { _, _ = reader.ExecContext(ctx, "ROLLBACK") }()
|
||||
|
||||
var routes int
|
||||
|
||||
err = reader.QueryRowContext(ctx,
|
||||
"SELECT count(*) FROM live_routes_v4").Scan(&routes)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to read the live routes: %v", err)
|
||||
}
|
||||
|
||||
err = db.UpsertLiveRouteBatch([]*LiveRoute{
|
||||
mkV4Route(t, "192.0.2.0/24", lockTestASN, time.Now().UTC()),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to store a route after the read: %v", err)
|
||||
}
|
||||
|
||||
_, lines := openLogged(t, cfg)
|
||||
line := findLine(t, lines, startupCheckpointIncompleteMessage)
|
||||
|
||||
if line[slog.LevelKey] != slog.LevelWarn.String() {
|
||||
t.Errorf("start-up checkpoint logged at level %v, want %s",
|
||||
line[slog.LevelKey], slog.LevelWarn)
|
||||
}
|
||||
|
||||
if line["busy"] != 1.0 {
|
||||
t.Errorf("start-up checkpoint logged busy = %v, want 1", line["busy"])
|
||||
}
|
||||
|
||||
walFrames, _ := line["wal_frames"].(float64)
|
||||
checkpointedFrames, _ := line["checkpointed_frames"].(float64)
|
||||
|
||||
if checkpointedFrames >= walFrames {
|
||||
t.Errorf("start-up checkpoint logged checkpointed_frames = %v and "+
|
||||
"wal_frames = %v with a read open, want fewer copied than the WAL "+
|
||||
"holds", line["checkpointed_frames"], line["wal_frames"])
|
||||
}
|
||||
}
|
||||
|
||||
// openLogged opens the database of cfg with a logger that writes JSON, and
|
||||
// returns it with the lines it logged while it opened, decoded.
|
||||
func openLogged(t *testing.T, cfg *config.Config) (*Database, []map[string]any) {
|
||||
t.Helper()
|
||||
|
||||
var logged bytes.Buffer
|
||||
|
||||
db, err := New(cfg,
|
||||
&logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logged, nil))})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to open the database: %v", err)
|
||||
}
|
||||
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
|
||||
var lines []map[string]any
|
||||
|
||||
decoder := json.NewDecoder(&logged)
|
||||
for decoder.More() {
|
||||
var line map[string]any
|
||||
|
||||
err := decoder.Decode(&line)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to decode the log: %v", err)
|
||||
}
|
||||
|
||||
lines = append(lines, line)
|
||||
}
|
||||
|
||||
return db, lines
|
||||
}
|
||||
|
||||
// findLine returns the first of lines logged with message.
|
||||
func findLine(
|
||||
t *testing.T,
|
||||
lines []map[string]any,
|
||||
message string,
|
||||
) map[string]any {
|
||||
t.Helper()
|
||||
|
||||
for _, line := range lines {
|
||||
if line[slog.MessageKey] == message {
|
||||
return line
|
||||
}
|
||||
}
|
||||
|
||||
t.Fatalf("no line %q in the log", message)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// TestGetRandomPrefixesByLengthListsPrefixOnce stores one prefix of each
|
||||
// family learned from two peers in the same second and checks that the
|
||||
// prefix-length sample lists each of them once.
|
||||
|
||||
@@ -1,61 +0,0 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.eeqj.de/sneak/routewatch/internal/config"
|
||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||
)
|
||||
|
||||
// TestDockerignoreExcludesDatabaseFiles checks that .dockerignore keeps the
|
||||
// database file, and the WAL and shared-memory files SQLite writes next to it,
|
||||
// out of the build context wherever they sit in the clone. The image ships the
|
||||
// whole build context as a source archive.
|
||||
func TestDockerignoreExcludesDatabaseFiles(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create database: %v", err)
|
||||
}
|
||||
defer func() { _ = db.Close() }()
|
||||
|
||||
data, err := os.ReadFile("../../.dockerignore")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to read .dockerignore: %v", err)
|
||||
}
|
||||
|
||||
// Docker reads `**/` as any number of directories and the rest of the
|
||||
// pattern as filepath.Match does, so `**/` followed by a pattern that
|
||||
// matches a file's name excludes that file at any depth.
|
||||
var anyDepth []string
|
||||
|
||||
for line := range strings.Lines(string(data)) {
|
||||
pattern, ok := strings.CutPrefix(strings.TrimSpace(line), "**/")
|
||||
if ok {
|
||||
anyDepth = append(anyDepth, pattern)
|
||||
}
|
||||
}
|
||||
|
||||
name := filepath.Base(db.path)
|
||||
|
||||
for _, file := range []string{name, name + "-wal", name + "-shm"} {
|
||||
excluded := false
|
||||
|
||||
for _, pattern := range anyDepth {
|
||||
matched, err := filepath.Match(pattern, file)
|
||||
if err != nil {
|
||||
t.Fatalf("bad .dockerignore pattern **/%s: %v", pattern, err)
|
||||
}
|
||||
|
||||
excluded = excluded || matched
|
||||
}
|
||||
|
||||
if !excluded {
|
||||
t.Errorf(".dockerignore has no **/ pattern that excludes %s", file)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -4,7 +4,6 @@ package ristypes
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"slices"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -25,8 +24,7 @@ const (
|
||||
)
|
||||
|
||||
// ASPath is a BGP AS path as the feed writes it: an array of AS numbers in
|
||||
// which an AS set is an array of its members, such as [1, 2, [3, 4]]. A path
|
||||
// that holds AS 0 decodes as an empty one (UnmarshalJSON).
|
||||
// which an AS set is an array of its members, such as [1, 2, [3, 4]].
|
||||
type ASPath struct {
|
||||
// Sequence is the path's AS numbers in order. A set of one member that
|
||||
// ends the path counts as that AS and is its last element. The members of
|
||||
@@ -50,41 +48,23 @@ func (p *ASPath) Origin() int {
|
||||
return p.Sequence[len(p.Sequence)-1]
|
||||
}
|
||||
|
||||
// UnmarshalJSON implements the json.Unmarshaler interface for ASPath. A path
|
||||
// that holds AS 0 anywhere, a member of an AS set included, decodes as an empty
|
||||
// path: RFC 7607 makes such a path malformed and its announcements
|
||||
// withdrawals. On an announcement with an empty path the prefix handler
|
||||
// deletes the peer's route, and the AS and peering handlers record nothing.
|
||||
// UnmarshalJSON implements the json.Unmarshaler interface for ASPath.
|
||||
func (p *ASPath) UnmarshalJSON(data []byte) error {
|
||||
path, err := decodeASPath(data)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if slices.Contains(path.Sequence, 0) || slices.Contains(path.Set, 0) {
|
||||
path = ASPath{}
|
||||
}
|
||||
|
||||
*p = path
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// decodeASPath decodes an AS path as the feed writes it.
|
||||
func decodeASPath(data []byte) (ASPath, error) {
|
||||
// Most paths have no AS set and decode as a plain array of integers.
|
||||
var sequence []int
|
||||
|
||||
err := json.Unmarshal(data, &sequence)
|
||||
if err == nil {
|
||||
return ASPath{Sequence: sequence}, nil
|
||||
*p = ASPath{Sequence: sequence}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
var raw []any
|
||||
|
||||
err = json.Unmarshal(data, &raw)
|
||||
if err != nil {
|
||||
return ASPath{}, err
|
||||
return err
|
||||
}
|
||||
|
||||
path := ASPath{Sequence: make([]int, 0, len(raw))}
|
||||
@@ -110,7 +90,9 @@ func decodeASPath(data []byte) (ASPath, error) {
|
||||
}
|
||||
}
|
||||
|
||||
return path, nil
|
||||
*p = path
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// RISLiveMessage represents the outer wrapper message from the RIPE RIS Live stream.
|
||||
|
||||
@@ -145,33 +145,3 @@ func TestASPathKeepsATrailingSetApart(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestASPathWithASZeroDecodesAsAnEmptyPath decodes AS paths that hold AS 0
|
||||
// before their end, at their end, or in an AS set. RFC 7607 makes such a path
|
||||
// malformed, so each decodes as an empty path.
|
||||
func TestASPathWithASZeroDecodesAsAnEmptyPath(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
for _, path := range []string{
|
||||
"[64496, 0, 64497]",
|
||||
"[64496, 0]",
|
||||
"[64496, [0, 64497], 64498]",
|
||||
"[64496, [64497, 0]]",
|
||||
} {
|
||||
t.Run(path, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var got ristypes.ASPath
|
||||
|
||||
err := json.Unmarshal([]byte(path), &got)
|
||||
if err != nil {
|
||||
t.Fatalf("unmarshal %s: %v", path, err)
|
||||
}
|
||||
|
||||
if len(got.Sequence) > 0 || len(got.Set) > 0 {
|
||||
t.Errorf("%s decoded to sequence %v and set %v, want an empty path",
|
||||
path, got.Sequence, got.Set)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -116,18 +116,17 @@ func (rw *RouteWatch) Run(ctx context.Context) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// Start ASN WHOIS fetcher for background updates. The HTTP handlers read
|
||||
// the server's fetcher without a lock, so it is set before they run.
|
||||
rw.asnFetcher = NewASNFetcher(rw.db, rw.logger.Logger)
|
||||
rw.asnFetcher.Start(ctx)
|
||||
rw.server.SetASNFetcher(rw.asnFetcher)
|
||||
|
||||
// Start HTTP server
|
||||
err = rw.server.Start(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Start ASN WHOIS fetcher for background updates
|
||||
rw.asnFetcher = NewASNFetcher(rw.db, rw.logger.Logger)
|
||||
rw.asnFetcher.Start(ctx)
|
||||
rw.server.SetASNFetcher(rw.asnFetcher)
|
||||
|
||||
// Start database maintenance goroutine
|
||||
rw.dbMaintainer = NewDBMaintainer(rw.db, rw.logger.Logger)
|
||||
rw.dbMaintainer.Start(ctx)
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
package routewatch
|
||||
|
||||
import (
|
||||
"maps"
|
||||
"slices"
|
||||
"testing"
|
||||
|
||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||
@@ -44,25 +42,3 @@ func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
|
||||
asn.LastSeen, second.ParsedTimestamp)
|
||||
}
|
||||
}
|
||||
|
||||
// TestASHandlerRecordsNothingFromAPathWithASZero handles an update whose AS
|
||||
// path holds AS 0 between two other ASes. RFC 7607 makes such a path
|
||||
// malformed, so neither AS is recorded.
|
||||
func TestASHandlerRecordsNothingFromAPathWithASZero(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
store := newMockStore()
|
||||
|
||||
// Built without NewASHandler's flush goroutine, so the test flushes itself.
|
||||
h := &ASHandler{db: store, logger: logger.New()}
|
||||
|
||||
msg := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||
msg.Path = decodePath(t, testASNA, 0, testASNB)
|
||||
h.HandleMessage(msg)
|
||||
h.flushBatch()
|
||||
|
||||
if len(store.ASNs) > 0 {
|
||||
t.Errorf("ASes recorded = %v, want none",
|
||||
slices.Sorted(maps.Keys(store.ASNs)))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -53,8 +53,7 @@ type ASNFetcher struct {
|
||||
currentInterval time.Duration
|
||||
consecutiveFails int
|
||||
|
||||
// hourly stats tracking; now is the clock, time.Now except in tests
|
||||
now func() time.Time
|
||||
// hourly stats tracking
|
||||
statsMu sync.Mutex
|
||||
successTimes []time.Time
|
||||
errorTimes []time.Time
|
||||
@@ -70,7 +69,6 @@ func NewASNFetcher(db database.Store, logger *slog.Logger) *ASNFetcher {
|
||||
stopCh: make(chan struct{}),
|
||||
done: make(chan struct{}),
|
||||
currentInterval: baseInterval,
|
||||
now: time.Now,
|
||||
successTimes: make([]time.Time, 0),
|
||||
errorTimes: make([]time.Time, 0),
|
||||
}
|
||||
@@ -123,7 +121,7 @@ func (f *ASNFetcher) GetStats() server.ASNFetcherStats {
|
||||
f.intervalMu.Unlock()
|
||||
|
||||
// Prune old entries and count
|
||||
cutoff := f.now().Add(-statsWindow)
|
||||
cutoff := time.Now().Add(-statsWindow)
|
||||
f.successTimes = pruneOldTimes(f.successTimes, cutoff)
|
||||
f.errorTimes = pruneOldTimes(f.errorTimes, cutoff)
|
||||
|
||||
@@ -135,15 +133,16 @@ func (f *ASNFetcher) GetStats() server.ASNFetcherStats {
|
||||
}
|
||||
}
|
||||
|
||||
// pruneOldTimes drops the times at or before cutoff from the front of times,
|
||||
// which holds times in the order they were recorded, and returns the rest. It
|
||||
// copies nothing, so it is cheap enough to run on every append.
|
||||
// pruneOldTimes removes times older than cutoff and returns the pruned slice.
|
||||
func pruneOldTimes(times []time.Time, cutoff time.Time) []time.Time {
|
||||
for len(times) > 0 && !times[0].After(cutoff) {
|
||||
times = times[1:]
|
||||
result := make([]time.Time, 0, len(times))
|
||||
for _, t := range times {
|
||||
if t.After(cutoff) {
|
||||
result = append(result, t)
|
||||
}
|
||||
}
|
||||
|
||||
return times
|
||||
return result
|
||||
}
|
||||
|
||||
// getInterval returns the current fetch interval.
|
||||
@@ -171,11 +170,9 @@ func (f *ASNFetcher) recordSuccess() {
|
||||
}
|
||||
f.intervalMu.Unlock()
|
||||
|
||||
// Record success time for stats, dropping those older than statsWindow
|
||||
// Record success time for stats
|
||||
f.statsMu.Lock()
|
||||
now := f.now()
|
||||
f.successTimes = pruneOldTimes(f.successTimes, now.Add(-statsWindow))
|
||||
f.successTimes = append(f.successTimes, now)
|
||||
f.successTimes = append(f.successTimes, time.Now())
|
||||
f.statsMu.Unlock()
|
||||
}
|
||||
|
||||
@@ -197,11 +194,9 @@ func (f *ASNFetcher) recordFailure() {
|
||||
}
|
||||
f.intervalMu.Unlock()
|
||||
|
||||
// Record error time for stats, dropping those older than statsWindow
|
||||
// Record error time for stats
|
||||
f.statsMu.Lock()
|
||||
now := f.now()
|
||||
f.errorTimes = pruneOldTimes(f.errorTimes, now.Add(-statsWindow))
|
||||
f.errorTimes = append(f.errorTimes, now)
|
||||
f.errorTimes = append(f.errorTimes, time.Now())
|
||||
f.statsMu.Unlock()
|
||||
}
|
||||
|
||||
|
||||
@@ -203,35 +203,3 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
|
||||
t.Fatal("the fetcher goroutine did not return after the lookup ended")
|
||||
}
|
||||
}
|
||||
|
||||
// TestRecordKeepsOnlyTheStatsWindow records a success and a failure every
|
||||
// second, by a clock the test sets, far longer than statsWindow, and never
|
||||
// asks for the stats. The fetcher must keep only the times within the last
|
||||
// statsWindow.
|
||||
func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
records = 100_000
|
||||
step = time.Second
|
||||
want = int(statsWindow / step)
|
||||
)
|
||||
|
||||
clock := time.Now()
|
||||
|
||||
f := NewASNFetcher(newMockStore(), logger.New().Logger)
|
||||
f.now = func() time.Time { return clock }
|
||||
|
||||
for range records {
|
||||
clock = clock.Add(step)
|
||||
|
||||
f.recordSuccess()
|
||||
f.recordFailure()
|
||||
}
|
||||
|
||||
if len(f.successTimes) != want || len(f.errorTimes) != want {
|
||||
t.Errorf("%d success and %d error times kept, want %d of each, "+
|
||||
"those of the last %s", len(f.successTimes), len(f.errorTimes),
|
||||
want, statsWindow)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -46,7 +46,7 @@ type PeeringHandler struct {
|
||||
// and the latest time an update carried it.
|
||||
mu sync.Mutex
|
||||
asPaths map[string]database.SeenTimes
|
||||
droppedPaths int // paths dropped since the last run because the map was full
|
||||
droppedPaths int // paths dropped because the map was full
|
||||
// writing is how many paths processPeerings took from asPaths and has
|
||||
// not finished writing the peerings of.
|
||||
writing int
|
||||
@@ -190,14 +190,12 @@ func (h *PeeringHandler) processLoop() {
|
||||
func (h *PeeringHandler) processPeerings() {
|
||||
// Take the accumulated paths and replace the map with a fresh empty one
|
||||
// under the lock. Each path is processed exactly once and the memory is
|
||||
// released, so the map never grows past a single interval's traffic. The
|
||||
// count of dropped paths starts again with the new map.
|
||||
// released, so the map never grows past a single interval's traffic.
|
||||
h.mu.Lock()
|
||||
paths := h.asPaths
|
||||
h.asPaths = make(map[string]database.SeenTimes)
|
||||
h.writing = len(paths)
|
||||
dropped := h.droppedPaths
|
||||
h.droppedPaths = 0
|
||||
h.mu.Unlock()
|
||||
|
||||
if len(paths) == 0 {
|
||||
|
||||
@@ -153,43 +153,6 @@ func TestPeeringHandlerLeavesATrailingASSetOutOfThePeerings(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestPeeringHandlerRecordsNothingFromAPathWithASZero handles AS paths that
|
||||
// hold AS 0. RFC 7607 makes such a path malformed, so no peering is recorded
|
||||
// from it, not even one between two other ASes of the path.
|
||||
func TestPeeringHandlerRecordsNothingFromAPathWithASZero(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
path []any
|
||||
}{
|
||||
{name: "AS 0 between two ASes", path: []any{testASNA, 0, testASNB}},
|
||||
{
|
||||
name: "AS 0 after two ASes",
|
||||
path: []any{testASNA, testASNB, 0, testASNC},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
store := newMockStore()
|
||||
h := newTestHandler(store)
|
||||
h.HandleMessage(&ristypes.RISMessage{
|
||||
Path: decodePath(t, tt.path...),
|
||||
ParsedTimestamp: time.Now().UTC(),
|
||||
})
|
||||
h.ProcessPeeringsNow()
|
||||
|
||||
if len(store.Peerings) > 0 {
|
||||
t.Errorf("peerings recorded = %v, want none",
|
||||
slices.Sorted(maps.Keys(store.Peerings)))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestPeeringHandlerProcessesEachRunAndEmpties verifies that a processing run
|
||||
// empties the path map (the swap) and that a path seen again after a run is
|
||||
// recorded in the next run too.
|
||||
@@ -301,36 +264,6 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestPeeringHandlerRunResetsTheDroppedCount fills the map, has the handler
|
||||
// drop one path, and runs the processing. The count of dropped paths is the
|
||||
// run's own, so it is 0 again after the run.
|
||||
func TestPeeringHandlerRunResetsTheDroppedCount(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
h := newTestHandler(newMockStore())
|
||||
|
||||
now := time.Now().UTC()
|
||||
|
||||
// Paths of one AS carry no peering, so the run records nothing.
|
||||
h.mu.Lock()
|
||||
for asn := 1; len(h.asPaths) < maxTrackedPaths; asn++ {
|
||||
h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now}
|
||||
}
|
||||
h.mu.Unlock()
|
||||
|
||||
handle(h, now, testASNA, testASNB)
|
||||
|
||||
if _, dropped := snapshot(h); dropped != 1 {
|
||||
t.Fatalf("want dropped count 1 with the map full, got %d", dropped)
|
||||
}
|
||||
|
||||
h.ProcessPeeringsNow()
|
||||
|
||||
if _, dropped := snapshot(h); dropped != 0 {
|
||||
t.Errorf("want dropped count 0 after a run, got %d", dropped)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPeeringHandlerStoresTheEarliestAndLatestTimesOfANewPeering handles two
|
||||
// updates whose path carries a peering the database does not hold yet, the
|
||||
// second 5 seconds after the first, in one run into a real database. The
|
||||
|
||||
@@ -45,8 +45,8 @@ const (
|
||||
|
||||
// PrefixHandler tracks BGP prefixes and maintains a live routing table in the
|
||||
// database. Routes are added on announcement and deleted on withdrawal or on an
|
||||
// announcement whose AS path is empty or holds AS 0 anywhere, and all routes of
|
||||
// a peer are deleted when its BGP session goes down.
|
||||
// announcement whose AS path is empty or ends in AS 0, and all routes of a peer
|
||||
// are deleted when its BGP session goes down.
|
||||
type PrefixHandler struct {
|
||||
db database.Store
|
||||
logger *logger.Logger
|
||||
@@ -373,10 +373,10 @@ func (h *PrefixHandler) flushBatch() {
|
||||
// An announcement is stored when its path gives an origin, or when it
|
||||
// ends in an AS set of several members: that route is stored with
|
||||
// origin 0 (none). Any other update deletes the peer's route for the
|
||||
// prefix: a withdrawal, or an announcement whose path is empty or
|
||||
// holds AS 0 anywhere (the decoder empties such a path). Such an
|
||||
// announcement replaces the peer's route, as any announcement does,
|
||||
// with none (RFC 7607 treats AS 0 in a path as a withdrawal).
|
||||
// prefix: a withdrawal, or an announcement whose path is empty or ends
|
||||
// in AS 0. Such an announcement replaces the peer's route, as any
|
||||
// announcement does, with none (RFC 7607 treats AS 0 in a path as a
|
||||
// withdrawal).
|
||||
if update.messageType == "announcement" &&
|
||||
(update.originASN > 0 || update.endsInSet) {
|
||||
// Create live route for batch upsert
|
||||
@@ -388,7 +388,7 @@ func (h *PrefixHandler) flushBatch() {
|
||||
// Parse CIDR to get IP version
|
||||
_, ipVersion, err := parseCIDR(update.prefix)
|
||||
if err != nil {
|
||||
h.logger.Error("Failed to parse CIDR for route deletion",
|
||||
h.logger.Error("Failed to parse CIDR for withdrawal",
|
||||
"prefix", update.prefix, "error", err)
|
||||
|
||||
continue
|
||||
|
||||
@@ -381,12 +381,11 @@ func TestPrefixHandlerDoesNotStoreAnAnnouncementWithAnEmptyPath(t *testing.T) {
|
||||
// TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin runs, through
|
||||
// the prefix handler into a real database, a peer's announcement of a prefix
|
||||
// with origin AS testASNB and then the peer's announcement of the prefix with
|
||||
// an AS path that gives no origin: one whose last AS is 0, an empty one, or one
|
||||
// that holds AS 0 before its end, which decodes as an empty path. As any later
|
||||
// announcement from the peer does, the second replaces the peer's route, here
|
||||
// with none, whether it comes in a later batch or follows a withdrawal in one
|
||||
// batch. The in-memory route count and prefix distribution must match the
|
||||
// route table.
|
||||
// an AS path that gives no origin: one whose last AS is 0, or an empty one. As
|
||||
// any later announcement from the peer does, the second replaces the peer's
|
||||
// route, here with none, whether it comes in a later batch or follows a
|
||||
// withdrawal in one batch. The in-memory route count and prefix distribution
|
||||
// must match the route table.
|
||||
func TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin(
|
||||
t *testing.T,
|
||||
) {
|
||||
@@ -422,12 +421,6 @@ func TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin(
|
||||
name: "an empty path",
|
||||
batch: []*ristypes.RISMessage{noOrigin(ristypes.ASPath{})},
|
||||
},
|
||||
{
|
||||
name: "a path with AS 0 before its end",
|
||||
batch: []*ristypes.RISMessage{
|
||||
noOrigin(decodePath(t, testASNA, 0, testASNB)),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "a withdrawal and then a path whose last AS is 0",
|
||||
batch: []*ristypes.RISMessage{withdrawal, noOrigin(endsInZero)},
|
||||
@@ -447,64 +440,6 @@ func TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin(
|
||||
}
|
||||
}
|
||||
|
||||
// TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse runs an announcement of
|
||||
// "not-a-prefix" and a withdrawal of "also/bad" through the prefix handler into
|
||||
// a real database. Neither is stored in the prefix tables: the prefix count in
|
||||
// the stats stays 0, both as kept in memory and as counted from the tables when
|
||||
// the database is opened again.
|
||||
func TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cfg := &config.Config{StateDir: t.TempDir()}
|
||||
|
||||
db, err := database.New(cfg, logger.New())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create database: %v", err)
|
||||
}
|
||||
|
||||
// Built without NewPrefixHandler's flush timer, so the test flushes itself.
|
||||
h := &PrefixHandler{db: db, logger: logger.New()}
|
||||
|
||||
withdrawal := &ristypes.RISMessage{
|
||||
Type: ristypes.MessageTypeUpdate,
|
||||
Peer: testPeerIP,
|
||||
ParsedTimestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
|
||||
Withdrawals: []string{"also/bad"},
|
||||
}
|
||||
h.HandleMessage(peerAnnouncement(testPeerIP, "not-a-prefix"))
|
||||
h.HandleMessage(withdrawal)
|
||||
h.flushBatch()
|
||||
|
||||
checkNoPrefixes := func(when string) {
|
||||
t.Helper()
|
||||
|
||||
stats, err := db.GetStatsContext(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("GetStatsContext %s: %v", when, err)
|
||||
}
|
||||
|
||||
if stats.Prefixes != 0 {
|
||||
t.Errorf("prefix count %s = %d, want 0", when, stats.Prefixes)
|
||||
}
|
||||
}
|
||||
|
||||
checkNoPrefixes("after the flush")
|
||||
|
||||
err = db.Close()
|
||||
if err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
|
||||
db, err = database.New(cfg, logger.New())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to open the database again: %v", err)
|
||||
}
|
||||
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
|
||||
checkNoPrefixes("after opening the database again")
|
||||
}
|
||||
|
||||
// TestStateMessagesGoToThePrefixHandlerNotThePeerHandler checks that STATE
|
||||
// messages go to the prefix handler, which acts on a down, and not to the peer
|
||||
// handler, which would record a peer whose session never comes up as active.
|
||||
|
||||
@@ -142,9 +142,8 @@ func deadlinePassed(ctx context.Context) bool {
|
||||
//
|
||||
// If a response is not valid JSON or is empty when JSON is expected, the middleware
|
||||
// returns a properly formatted JSON error response. If the request's time limit
|
||||
// has passed, it answers 504 with the error message "Request timeout". A 4xx
|
||||
// keeps its status, with the status's standard text as the message. For other
|
||||
// errors, it returns "Internal server error", with status 500 if the original
|
||||
// has passed, it answers 504 with the error message "Request timeout". For other
|
||||
// errors, it returns "Internal server error" with status 500 if the original
|
||||
// status was 200.
|
||||
//
|
||||
// Non-JSON responses are passed through unchanged.
|
||||
@@ -204,13 +203,8 @@ func JSONValidationMiddleware(next http.Handler) http.Handler {
|
||||
w.WriteHeader(statusCode)
|
||||
|
||||
errorMsg := internalErrorMessage
|
||||
|
||||
switch {
|
||||
case statusCode == http.StatusGatewayTimeout:
|
||||
if statusCode == http.StatusGatewayTimeout {
|
||||
errorMsg = requestTimeoutMessage
|
||||
case statusCode >= http.StatusBadRequest &&
|
||||
statusCode < http.StatusInternalServerError:
|
||||
errorMsg = http.StatusText(statusCode)
|
||||
}
|
||||
|
||||
response := map[string]any{
|
||||
|
||||
@@ -58,34 +58,6 @@ func TestResponsesCarrySecurityHeaders(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestJSONValidationKeeps4xxWithoutBody wraps a handler that answers 400
|
||||
// without a body in JSONValidationMiddleware and checks that the JSON error
|
||||
// body keeps the 400 and gives the status's own text. Before the fix it said
|
||||
// "Internal server error"
|
||||
// (https://git.eeqj.de/sneak/routewatch/issues/111).
|
||||
func TestJSONValidationKeeps4xxWithoutBody(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
handler := JSONValidationMiddleware(http.HandlerFunc(
|
||||
func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
}))
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
handler.ServeHTTP(rec, httptest.NewRequestWithContext(
|
||||
t.Context(), http.MethodGet, "/", nil))
|
||||
|
||||
const name = "400 without a body"
|
||||
|
||||
if rec.Code != http.StatusBadRequest {
|
||||
t.Errorf("%s: status %d, want %d; body %s",
|
||||
name, rec.Code, http.StatusBadRequest, rec.Body)
|
||||
}
|
||||
|
||||
checkJSONErrorCode(t, name, rec, http.StatusBadRequest)
|
||||
checkJSONErrorMessage(t, name, rec, http.StatusText(http.StatusBadRequest))
|
||||
}
|
||||
|
||||
// TestPagesCarryTheCSPNonce checks that every script and style block of each
|
||||
// page carries the nonce its Content-Security-Policy header names, so that the
|
||||
// browser runs and applies them, that each answer has a nonce of its own, and
|
||||
|
||||
@@ -1,33 +1,18 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/go-chi/chi/v5/middleware"
|
||||
)
|
||||
|
||||
// The error messages the JSON routes under /api/v1 and /ip answer with when no
|
||||
// route matches the path, and when none takes the method.
|
||||
const (
|
||||
notFoundMessage = "Not found"
|
||||
methodNotAllowedMessage = "Method not allowed"
|
||||
|
||||
// jsonRoutesAllow is the Allow header of their 405 answer: every one of
|
||||
// them takes GET and HEAD, and no other method.
|
||||
jsonRoutesAllow = "GET, HEAD"
|
||||
)
|
||||
|
||||
// setupRoutes configures the HTTP routes
|
||||
func (s *Server) setupRoutes() {
|
||||
r := chi.NewRouter()
|
||||
|
||||
// Middleware
|
||||
r.Use(DecodedPathMiddleware)
|
||||
// GetHead serves HEAD with the GET handler. It looks the route up by the
|
||||
// path DecodedPathMiddleware sets, so it comes after it.
|
||||
r.Use(middleware.GetHead)
|
||||
r.Use(SecurityHeadersMiddleware)
|
||||
r.Use(middleware.RequestID)
|
||||
r.Use(RealIPMiddleware(s.trustedProxies))
|
||||
@@ -55,34 +40,16 @@ func (s *Server) setupRoutes() {
|
||||
r.Get("/prefixlength/{length}", s.handlePrefixLength())
|
||||
r.Get("/prefixlength6/{length}", s.handlePrefixLength6())
|
||||
|
||||
// The JSON routes answer a path or method they do not take with the JSON
|
||||
// error body, as they answer everything else.
|
||||
jsonNotFound := func(w http.ResponseWriter, _ *http.Request) {
|
||||
s.writeJSONError(w, http.StatusNotFound, notFoundMessage)
|
||||
}
|
||||
jsonMethodNotAllowed := func(w http.ResponseWriter, _ *http.Request) {
|
||||
// chi's own 405 answer sets Allow; one set in its place must too.
|
||||
w.Header().Set("Allow", jsonRoutesAllow)
|
||||
s.writeJSONError(w, http.StatusMethodNotAllowed, methodNotAllowedMessage)
|
||||
}
|
||||
|
||||
// IP info JSON endpoints (replaces old /ip redirect)
|
||||
r.Route("/ip", func(r chi.Router) {
|
||||
r.Use(JSONValidationMiddleware)
|
||||
r.NotFound(jsonNotFound)
|
||||
r.MethodNotAllowed(jsonMethodNotAllowed)
|
||||
r.Get("/", s.handleIPInfo()) // Client IP
|
||||
// GetHead does not reach this route: mounting the group registers
|
||||
// /ip and /ip/ for every method, so HEAD arrives here as HEAD.
|
||||
r.Head("/", s.handleIPInfo())
|
||||
r.Get("/{addr}", s.handleIPInfo()) // Specified IP
|
||||
})
|
||||
|
||||
// API routes
|
||||
r.Route("/api/v1", func(r chi.Router) {
|
||||
r.Use(JSONValidationMiddleware)
|
||||
r.NotFound(jsonNotFound)
|
||||
r.MethodNotAllowed(jsonMethodNotAllowed)
|
||||
r.Get("/stats", s.handleStats())
|
||||
r.Get("/ip/{ip}", s.handleIPLookup())
|
||||
r.Get("/as/{asn}", s.handleASDetailJSON())
|
||||
|
||||
@@ -1,100 +0,0 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestRoutesAnswerHEAD sends HEAD for a page and JSON routes through the
|
||||
// server's router and checks that each answers as GET does. /ip and /ip/, the
|
||||
// lookup of the caller's own address, are in the list because chi's GetHead
|
||||
// does not reach them. Before the fix every route answered HEAD with 405
|
||||
// (https://git.eeqj.de/sneak/routewatch/issues/111).
|
||||
func TestRoutesAnswerHEAD(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newServerWithTestRoutes(t)
|
||||
|
||||
for _, path := range []string{"/", statsPath, "/ip", "/ip/"} {
|
||||
rec := httptest.NewRecorder()
|
||||
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
|
||||
t.Context(), http.MethodHead, path, nil))
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Errorf("HEAD %s: status %d, want %d; body %s",
|
||||
path, rec.Code, http.StatusOK, rec.Body)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestJSONRoutesAnswerUnmatchedRequestsWithJSONErrors sends a method the JSON
|
||||
// routes do not take, and a path none of them matches, to each group of JSON
|
||||
// routes through the server's router. Each must answer with the JSON error
|
||||
// body, and a 405 must name the methods the routes take in its Allow header.
|
||||
// Before the fix a 405 said "Internal server error" and a 404 was chi's plain
|
||||
// text answer (https://git.eeqj.de/sneak/routewatch/issues/111).
|
||||
func TestJSONRoutesAnswerUnmatchedRequestsWithJSONErrors(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newServerWithTestRoutes(t)
|
||||
|
||||
tests := []struct {
|
||||
method string
|
||||
path string
|
||||
wantCode int
|
||||
wantMsg string
|
||||
}{
|
||||
{http.MethodPost, statsPath, http.StatusMethodNotAllowed, methodNotAllowedMessage},
|
||||
{http.MethodPost, "/ip/", http.StatusMethodNotAllowed, methodNotAllowedMessage},
|
||||
{http.MethodGet, "/api/v1/nothing", http.StatusNotFound, notFoundMessage},
|
||||
{http.MethodGet, "/ip/192.0.2.1/nothing", http.StatusNotFound, notFoundMessage},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
name := tt.method + " " + tt.path
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
|
||||
t.Context(), tt.method, tt.path, nil))
|
||||
|
||||
if rec.Code != tt.wantCode {
|
||||
t.Errorf("%s: status %d, want %d; body %s",
|
||||
name, rec.Code, tt.wantCode, rec.Body)
|
||||
}
|
||||
|
||||
checkJSONErrorCode(t, name, rec, tt.wantCode)
|
||||
checkJSONErrorMessage(t, name, rec, tt.wantMsg)
|
||||
|
||||
allow := rec.Header().Get("Allow")
|
||||
if tt.wantCode == http.StatusMethodNotAllowed && allow != jsonRoutesAllow {
|
||||
t.Errorf("%s: Allow %q, want %q", name, allow, jsonRoutesAllow)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// checkJSONErrorMessage checks that rec holds the JSON error body with the
|
||||
// given message.
|
||||
func checkJSONErrorMessage(
|
||||
t *testing.T, name string, rec *httptest.ResponseRecorder, want string,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
var body struct {
|
||||
Error struct {
|
||||
Msg string `json:"msg"`
|
||||
} `json:"error"`
|
||||
}
|
||||
|
||||
err := json.Unmarshal(rec.Body.Bytes(), &body)
|
||||
if err != nil {
|
||||
t.Errorf("%s: decoding the answer %q: %v", name, rec.Body, err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if body.Error.Msg != want {
|
||||
t.Errorf("%s: error message %q, want %q", name, body.Error.Msg, want)
|
||||
}
|
||||
}
|
||||
@@ -139,8 +139,7 @@ func (s *Server) Stop(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetASNFetcher sets the ASN WHOIS fetcher for on-demand lookups. Call it
|
||||
// before Start: the handlers read the fetcher without a lock.
|
||||
// SetASNFetcher sets the ASN WHOIS fetcher for on-demand lookups.
|
||||
func (s *Server) SetASNFetcher(fetcher ASNFetcher) {
|
||||
s.asnFetcher = fetcher
|
||||
}
|
||||
|
||||
@@ -106,12 +106,12 @@ type RawMessageHandler func(line string)
|
||||
// handlerMetrics tracks performance metrics for a handler
|
||||
type handlerMetrics struct {
|
||||
processedCount uint64 // Total messages processed
|
||||
droppedCount atomic.Uint64 // Total messages dropped; not under mu
|
||||
droppedCount uint64 // Total messages dropped
|
||||
totalTime time.Duration // Total processing time (for average calculation)
|
||||
minTime time.Duration // Minimum processing time
|
||||
maxTime time.Duration // Maximum processing time
|
||||
queueHighWaterMark int // Maximum queue length seen
|
||||
mu sync.Mutex // Protects the other metrics
|
||||
mu sync.Mutex // Protects the metrics
|
||||
}
|
||||
|
||||
// handlerInfo wraps a handler with its queue and metrics
|
||||
@@ -371,7 +371,7 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
|
||||
QueueCapacity: cap(info.queue),
|
||||
QueueHighWaterMark: info.metrics.queueHighWaterMark,
|
||||
ProcessedCount: info.metrics.processedCount,
|
||||
DroppedCount: info.metrics.droppedCount.Load(),
|
||||
DroppedCount: info.metrics.droppedCount,
|
||||
MinProcessTime: info.metrics.minTime,
|
||||
MaxProcessTime: info.metrics.maxTime,
|
||||
}
|
||||
@@ -488,7 +488,7 @@ func (s *Streamer) logMetrics() {
|
||||
"queue_len", len(info.queue),
|
||||
"queue_cap", cap(info.queue),
|
||||
"processed", info.metrics.processedCount,
|
||||
"dropped", info.metrics.droppedCount.Load(),
|
||||
"dropped", info.metrics.droppedCount,
|
||||
"avg_time", avgTime,
|
||||
"min_time", info.metrics.minTime,
|
||||
"max_time", info.metrics.maxTime,
|
||||
@@ -858,7 +858,7 @@ func (s *Streamer) dispatchMessage(
|
||||
}
|
||||
|
||||
if !s.queueMessage(info, msg) {
|
||||
info.metrics.droppedCount.Add(1)
|
||||
atomic.AddUint64(&info.metrics.droppedCount, 1)
|
||||
s.totalDropped.Add(1)
|
||||
|
||||
continue
|
||||
|
||||
@@ -621,68 +621,6 @@ func TestStreamReadsLongLine(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// blockedHandler wants UPDATE messages, has room on its queue for one, and
|
||||
// does not return from HandleMessage until release is closed.
|
||||
type blockedHandler struct {
|
||||
release chan struct{}
|
||||
}
|
||||
|
||||
func (blockedHandler) WantsMessage(messageType string) bool {
|
||||
return messageType == ristypes.MessageTypeUpdate
|
||||
}
|
||||
|
||||
func (blockedHandler) QueueCapacity() int { return 1 }
|
||||
|
||||
func (h blockedHandler) HandleMessage(*ristypes.RISMessage) { <-h.release }
|
||||
|
||||
// TestHandlerStatsReadWhileMessagesAreDropped has the server send a few UPDATE
|
||||
// lines to a blockedHandler, so the read loop drops all but the one the
|
||||
// handler holds and the one on its queue, while the test reads the handler's
|
||||
// statistics. The drop count must reach the number dropped, and the race
|
||||
// detector must find the count read and written safely.
|
||||
func TestHandlerStatsReadWhileMessagesAreDropped(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
lines = 5
|
||||
wantDropped = lines - 2
|
||||
maxWait = 5 * time.Second
|
||||
)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(
|
||||
func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = io.WriteString(w, strings.Repeat(updateLine, lines))
|
||||
|
||||
flush(w)
|
||||
<-r.Context().Done()
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
s := New(logger.New(), metrics.New())
|
||||
s.url = srv.URL
|
||||
handler := blockedHandler{release: make(chan struct{})}
|
||||
s.RegisterHandler(handler)
|
||||
|
||||
err := s.Start(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("Start returned error: %v", err)
|
||||
}
|
||||
defer s.Stop(t.Context())
|
||||
defer close(handler.release)
|
||||
|
||||
// The test polls instead of waiting for a signal from the read loop: a
|
||||
// signal would order the drops before the reads here, and the race
|
||||
// detector would have nothing to report.
|
||||
deadline := time.Now().Add(maxWait)
|
||||
|
||||
for s.GetHandlerStats()[0].DroppedCount < wantDropped {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("fewer than %d messages counted as dropped within %s",
|
||||
wantDropped, maxWait)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestPeerCountCountsEveryPeerHeardFrom has the server send an UPDATE from
|
||||
// stateDownPeer and a KEEPALIVE from another peer, with no OPEN from either,
|
||||
// and then a message that ends stateDownPeer's session: a NOTIFICATION, or a
|
||||
|
||||
@@ -1,54 +0,0 @@
|
||||
package templates_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"html/template"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.eeqj.de/sneak/routewatch/internal/templates"
|
||||
)
|
||||
|
||||
// TestFooterLicense renders the home and status pages and expects each footer
|
||||
// to name the licence in LICENSE, MIT, and not WTFPL.
|
||||
func TestFooterLicense(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
license = "MIT"
|
||||
wrongLicense = "WTFPL"
|
||||
)
|
||||
|
||||
pages := map[string]*template.Template{
|
||||
"index": templates.IndexTemplate(),
|
||||
"status": templates.StatusTemplate(),
|
||||
}
|
||||
|
||||
for name, tmpl := range pages {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var page bytes.Buffer
|
||||
|
||||
err := tmpl.Execute(&page, struct{ CSPNonce string }{})
|
||||
if err != nil {
|
||||
t.Fatalf("rendering the page: %v", err)
|
||||
}
|
||||
|
||||
_, footer, found := strings.Cut(page.String(), "<footer")
|
||||
if !found {
|
||||
t.Fatal("the page has no footer")
|
||||
}
|
||||
|
||||
footer, _, _ = strings.Cut(footer, "</footer>")
|
||||
|
||||
if !strings.Contains(footer, license) {
|
||||
t.Errorf("footer does not name %s: %q", license, footer)
|
||||
}
|
||||
|
||||
if strings.Contains(footer, wrongLicense) {
|
||||
t.Errorf("footer names %s: %q", wrongLicense, footer)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -19,8 +19,8 @@ const (
|
||||
Author = "@sneak"
|
||||
// AuthorURL is the author's website
|
||||
AuthorURL = "https://sneak.berlin"
|
||||
// License is the program license, as LICENSE grants it
|
||||
License = "MIT"
|
||||
// License is the program license
|
||||
License = "WTFPL"
|
||||
// RepoURL is the git repository URL
|
||||
RepoURL = "https://git.eeqj.de/sneak/routewatch"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user