Merge feat/fed-reports-metrics
# Conflicts: # Sources/FlamenetRelay/Config.swift # Sources/FlamenetRelay/Migrations.swift
This commit is contained in:
+4
-2
@@ -54,7 +54,7 @@ FN_CONTACT_EMAIL=
|
||||
# FN_CONTACT_LEGAL=
|
||||
# FN_CONTACT_POSTMASTER=
|
||||
|
||||
# Where a report from inside the app is mailed. Needs the SMTP settings below.
|
||||
# Where the hourly digest of reports is mailed. Needs the SMTP settings below.
|
||||
# Blank, reports are still kept in the database and read with
|
||||
# Scripts/reports.sh, and the log says nobody was told. This is also the
|
||||
# fallback for FN_CONTACT_EMAIL when that is blank.
|
||||
@@ -184,7 +184,9 @@ FN_PUBLIC_HOST=
|
||||
FN_ROSTER_ON_SERVER=true
|
||||
|
||||
FN_FEDERATION=closed
|
||||
# Comma-separated peer hostnames, e.g.: relay.flamenet.io,relay.example.org
|
||||
# Comma-separated peers. Pin each one's federation key: host=key, or
|
||||
# host=key1|key2 during a rotation. A bare host is unpinned and logs a warning.
|
||||
# relay.flamenet.io=<base64 key>,relay.example.org=<base64 key>
|
||||
FN_FEDERATION_PEERS=
|
||||
|
||||
# This relay's federation signing identity, a 32-byte Ed25519 seed:
|
||||
|
||||
@@ -27,10 +27,13 @@ see `.env.example`.
|
||||
swift build -c release
|
||||
FN_ISSUER_PUBLIC_KEY=<base64 Ed25519 public key> \
|
||||
FN_RELAY_AUDIENCE=https://relay.flamenet.io \
|
||||
FN_DB=/var/lib/flamenet/relay.sqlite \
|
||||
FN_POSTGRES_URL=postgres://flamenet:<password>@localhost:5432/relay \
|
||||
.build/release/flamenet-relay
|
||||
```
|
||||
|
||||
Self-hosting requires Postgres. Without `FN_POSTGRES_URL` the relay uses a
|
||||
SQLite file, which is for local development only and receives no new features.
|
||||
|
||||
It refuses to start without an issuer key. There is no safe default for that
|
||||
value, and a relay that accepts unsigned tokens is worse than one that is down.
|
||||
|
||||
@@ -38,11 +41,14 @@ value, and a relay that accepts unsigned tokens is worse than one that is down.
|
||||
|---|---|---|
|
||||
| `FN_ISSUER_PUBLIC_KEY` | n/a | Required. Base64 Ed25519 public key of the token issuer. |
|
||||
| `FN_RELAY_AUDIENCE` | n/a | Required. The `aud` this relay accepts. |
|
||||
| `FN_DB` | `relay.sqlite` | SQLite file. |
|
||||
| `FN_POSTGRES_URL` | unset | Postgres. Required for self-hosting. |
|
||||
| `FN_DB` | `relay.sqlite` | SQLite file, used without `FN_POSTGRES_URL`. Local development only. |
|
||||
| `FN_HOST` / `FN_PORT` | `127.0.0.1` / `8080` | Bind address. |
|
||||
| `FN_WORKDIR` | inferred | Directory containing `Resources/` and `Public/`. Set it in a unit file. |
|
||||
| `FN_CLOCK_SKEW` | `60` | Seconds of tolerance on a proof timestamp. |
|
||||
| `FN_FEDERATION` | `closed` | `closed`, `allowlist` or `open`. Closed is a complete relay, not a limited one, and is what a self-hoster gets without configuring anything. An unrecognised value falls back to `closed` rather than refusing to boot, a typo in an optional variable should not stop a relay serving its own users. |
|
||||
| `FN_FEDERATION_PEERS` | empty | `host`, `host=key` or `host=key1\|key2`. A pinned peer publishing another key is refused. See `docs/CONFIGURATION.md` for rotation. |
|
||||
| `FN_METRICS_PORT` | unset | Prometheus `/metrics` on a second port, loopback by default. Off unless set. |
|
||||
| `FN_BUNDLE_RATE` / `FN_BUNDLE_RATE_TARGET` | `600` / `60` | Bundle fetches per hour, per requester and per pair. |
|
||||
| `FN_SEND_RATE` | `3000` | Envelopes per hour. |
|
||||
| `FN_DELIVERED_RETENTION_DAYS` | `1` | The backstop for a collected envelope the app never confirms. |
|
||||
|
||||
@@ -44,6 +44,11 @@ long, rendered from that instance's own configuration.
|
||||
throttling and sign-up limits apply to everyone at once.
|
||||
- Keep `FN_ISSUER_PRIVATE_KEY` and `FN_AT_REST_SECRET` out of the image and off
|
||||
shared disks. A leaked issuer key lets someone mint tokens for any account.
|
||||
- Pin every federation peer's key in `FN_FEDERATION_PEERS` (`host=key`).
|
||||
An unpinned peer is believed with whatever key its descriptor serves, and
|
||||
boot logs a warning for each one.
|
||||
- Keep `FN_METRICS_PORT` off the public network. The counters carry no ids,
|
||||
and they still describe your traffic.
|
||||
- Raise the file-descriptor limit (`ulimits: nofile: 65536`). Each held
|
||||
long-poll costs two; the default 1024 caps you near 450 open apps.
|
||||
- Back up the database *and* rehearse the restore. `docs/CAPACITY.md` has the
|
||||
|
||||
@@ -109,6 +109,7 @@ struct APNs: Sendable {
|
||||
let jwt: String
|
||||
do { jwt = try await Self.cache.current { try mintJWT() } } catch {
|
||||
logger.error("APNs token could not be minted: \(error)")
|
||||
RelayMetrics.shared.add(.pushFailed)
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -134,11 +135,14 @@ struct APNs: Sendable {
|
||||
switch response.status.code {
|
||||
case 200:
|
||||
await Self.health.succeeded()
|
||||
RelayMetrics.shared.add(.pushSent)
|
||||
return false
|
||||
case 410:
|
||||
RelayMetrics.shared.add(.pushFailed)
|
||||
logger.notice("APNs reports a device is gone; forgetting its token")
|
||||
return true
|
||||
default:
|
||||
RelayMetrics.shared.add(.pushFailed)
|
||||
let reason = response.body.map { String(buffer: $0) } ?? ""
|
||||
let proven = await Self.health.everWorked
|
||||
logger.error(
|
||||
@@ -151,6 +155,7 @@ struct APNs: Sendable {
|
||||
}
|
||||
} catch {
|
||||
// Never propagates.
|
||||
RelayMetrics.shared.add(.pushFailed)
|
||||
logger.error("APNs unreachable: \(error)")
|
||||
return false
|
||||
}
|
||||
@@ -162,6 +167,7 @@ struct APNs: Sendable {
|
||||
let jwt: String
|
||||
do { jwt = try await Self.cache.current { try mintJWT() } } catch {
|
||||
logger.error("APNs token could not be minted: \(error)")
|
||||
RelayMetrics.shared.add(.pushFailed)
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -184,8 +190,10 @@ struct APNs: Sendable {
|
||||
switch response.status.code {
|
||||
case 200:
|
||||
await Self.health.succeeded()
|
||||
RelayMetrics.shared.add(.pushSent)
|
||||
return false
|
||||
case 410:
|
||||
RelayMetrics.shared.add(.pushFailed)
|
||||
logger.notice("APNs reports a VoIP token is gone; forgetting it")
|
||||
return true
|
||||
default:
|
||||
@@ -193,6 +201,7 @@ struct APNs: Sendable {
|
||||
// the environment said only "failed: 400", the one path where
|
||||
// the diagnosis matters most, because a call that never rings
|
||||
// is what somebody notices first.
|
||||
RelayMetrics.shared.add(.pushFailed)
|
||||
let reason = response.body.map { String(buffer: $0) } ?? ""
|
||||
let proven = await Self.health.everWorked
|
||||
logger.error(
|
||||
@@ -207,6 +216,7 @@ struct APNs: Sendable {
|
||||
return proven && reason.contains("BadDeviceToken")
|
||||
}
|
||||
} catch {
|
||||
RelayMetrics.shared.add(.pushFailed)
|
||||
logger.error("APNs VoIP push could not be sent: \(error)")
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -88,6 +88,14 @@ struct Config: Sendable {
|
||||
/// The `iss` an account-service token must carry (`FN_ACCOUNT_TOKEN_ISSUER`).
|
||||
var accountTokenIssuer: String = ""
|
||||
|
||||
/// Content reports one account may file per hour (`FN_REPORT_RATE`).
|
||||
var reportRatePerHour: Int = 10
|
||||
|
||||
/// The Prometheus listener: a second port, never the public one. 0, the
|
||||
/// default, starts no listener. `FN_METRICS_HOST` defaults to loopback.
|
||||
var metricsHost: String = "127.0.0.1"
|
||||
var metricsPort: Int = 0
|
||||
|
||||
/// Whether this relay will exchange traffic with other relays.
|
||||
let federationPolicy: String
|
||||
|
||||
@@ -96,6 +104,11 @@ struct Config: Sendable {
|
||||
|
||||
/// Peers this relay will exchange traffic with under `allowlist`.
|
||||
let federationPeers: [String]
|
||||
/// The federation keys pinned for each peer, from `host=key` or
|
||||
/// `host=key1|key2` in `FN_FEDERATION_PEERS`. A peer with an entry here is
|
||||
/// refused when its descriptor publishes any other key. A peer without one
|
||||
/// is unpinned, and boot logs a warning for it.
|
||||
var federationPeerKeys: [String: [String]] = [:]
|
||||
/// Whether to screen the allowlist at boot and log what was seen
|
||||
/// (relay#58).
|
||||
let iosLive: Bool
|
||||
@@ -248,6 +261,7 @@ struct Config: Sendable {
|
||||
return raw == "1" || raw == "true" || raw == "yes" || raw == "on"
|
||||
}
|
||||
|
||||
let peers = parseFederationPeers(env["FN_FEDERATION_PEERS"] ?? "")
|
||||
var config = Config(
|
||||
host: env["FN_HOST"] ?? "127.0.0.1",
|
||||
port: int("FN_PORT", 8080),
|
||||
@@ -274,10 +288,7 @@ struct Config: Sendable {
|
||||
}(),
|
||||
federationKeyRaw: env["FN_FEDERATION_KEY"]
|
||||
.flatMap { $0.isEmpty ? nil : Data(base64Encoded: $0) },
|
||||
federationPeers: (env["FN_FEDERATION_PEERS"] ?? "")
|
||||
.split(separator: ",")
|
||||
.map { $0.trimmingCharacters(in: .whitespaces) }
|
||||
.filter { !$0.isEmpty },
|
||||
federationPeers: peers.hosts,
|
||||
iosLive: bool("FN_IOS_LIVE", false),
|
||||
iosStoreURL: env["FN_IOS_STORE_URL"]?.trimmingCharacters(in: .whitespaces) ?? "",
|
||||
federationScreening: {
|
||||
@@ -368,6 +379,10 @@ struct Config: Sendable {
|
||||
.flatMap { $0.isEmpty ? nil : Data(base64Encoded: $0) },
|
||||
tokenLifetime: int("FN_TOKEN_LIFETIME", Issuer.defaultLifetime)
|
||||
)
|
||||
config.federationPeerKeys = peers.keys
|
||||
config.reportRatePerHour = int("FN_REPORT_RATE", 10)
|
||||
config.metricsHost = env["FN_METRICS_HOST"].flatMap { $0.isEmpty ? nil : $0 } ?? "127.0.0.1"
|
||||
config.metricsPort = max(0, int("FN_METRICS_PORT", 0))
|
||||
config.padEnforce = PadEnforcement(setting: env["FN_PAD_ENFORCE"])
|
||||
config.padAttachments = bool("FN_PAD_ATTACHMENTS", false)
|
||||
config.introRatePerIP = int("FN_INTRO_RATE", 30)
|
||||
@@ -384,6 +399,52 @@ struct Config: Sendable {
|
||||
return config
|
||||
}
|
||||
|
||||
/// `host`, `host=key` or `host=key1|key2`, comma separated. A key is
|
||||
/// base64 and may end in `=`, so only the first `=` splits.
|
||||
static func parseFederationPeers(_ raw: String) -> (hosts: [String], keys: [String: [String]]) {
|
||||
var hosts: [String] = []
|
||||
var keys: [String: [String]] = [:]
|
||||
for entry in raw.split(separator: ",") {
|
||||
let parts = entry.split(separator: "=", maxSplits: 1)
|
||||
let host = parts.first.map { $0.trimmingCharacters(in: .whitespaces) } ?? ""
|
||||
guard !host.isEmpty else { continue }
|
||||
if !hosts.contains(host) { hosts.append(host) }
|
||||
guard parts.count == 2 else { continue }
|
||||
let pins = parts[1].split(separator: "|")
|
||||
.map { $0.trimmingCharacters(in: .whitespaces) }
|
||||
.filter { !$0.isEmpty }
|
||||
// `host=` with nothing after it pins an empty set, which refuses
|
||||
// every key: a pin the operator started typing fails closed.
|
||||
keys[host, default: []].append(contentsOf: pins)
|
||||
}
|
||||
return (hosts, keys)
|
||||
}
|
||||
|
||||
/// What boot logs about the pins: one line per unpinned peer, and one per
|
||||
/// pinned value that is not a 32-byte Ed25519 public key (it matches no
|
||||
/// descriptor, so that peer is refused until it is fixed).
|
||||
func federationPinWarnings() -> [String] {
|
||||
guard federationPolicy != "closed" else { return [] }
|
||||
var lines: [String] = []
|
||||
for host in federationPeers {
|
||||
guard let pins = federationPeerKeys[host] else {
|
||||
lines.append(
|
||||
"federation peer \(host) is unpinned: any key its descriptor publishes is "
|
||||
+ "accepted. Pin it as \(host)=<base64 key> in FN_FEDERATION_PEERS.")
|
||||
continue
|
||||
}
|
||||
if pins.isEmpty {
|
||||
lines.append("federation peer \(host) has an empty pin and is refused")
|
||||
}
|
||||
for pin in pins where Data(base64Encoded: pin)?.count != 32 {
|
||||
lines.append(
|
||||
"federation peer \(host) has a pinned key that is not a base64 32-byte "
|
||||
+ "Ed25519 public key; it matches nothing")
|
||||
}
|
||||
}
|
||||
return lines
|
||||
}
|
||||
|
||||
/// Where attachments actually live: the configured value if it is
|
||||
/// absolute, otherwise relative to the working directory.
|
||||
func attachmentsPath(relativeTo workingDirectory: String) -> String {
|
||||
|
||||
@@ -44,6 +44,7 @@ enum Delivery {
|
||||
let ids = try await insert(items, on: tx)
|
||||
return (ids, Set(items.map(\.toDevice)), result)
|
||||
}
|
||||
RelayMetrics.shared.add(.envelopesWritten, ids.count)
|
||||
// After the commit, never inside the transaction.
|
||||
if !ids.isEmpty { await notify(devices, push: push, on: app) }
|
||||
return (ids, result)
|
||||
|
||||
@@ -237,7 +237,8 @@ func configure(_ app: Application) async throws {
|
||||
audience: config.audience,
|
||||
skew: config.clockSkew,
|
||||
directory: peerDirectory,
|
||||
replay: replayCache),
|
||||
replay: replayCache,
|
||||
peerKeys: config.federationPeerKeys),
|
||||
federationSender: {
|
||||
guard config.federationPolicy != "closed",
|
||||
let seed = config.federationKeyRaw
|
||||
@@ -282,6 +283,7 @@ func configure(_ app: Application) async throws {
|
||||
"FN_FEDERATION=allowlist with no FN_FEDERATION_PEERS: this relay federates with nobody"
|
||||
)
|
||||
}
|
||||
for line in config.federationPinWarnings() { app.logger.warning("\(line)") }
|
||||
// relay#58, screen the allowlist and say what was seen.
|
||||
if config.federationScreening {
|
||||
let peers = config.federationPeers
|
||||
@@ -424,6 +426,26 @@ func configure(_ app: Application) async throws {
|
||||
|
||||
app.lifecycle.use(BootTask { await retention.run() })
|
||||
|
||||
// One mail per hour at most, listing the reports filed since the last.
|
||||
if config.contentReportsEnabled, let to = config.reportsTo, let mailer = routes.mailer {
|
||||
let digest = ReportDigest(
|
||||
db: app.db, to: to,
|
||||
send: { try await mailer.send(to: $0, subject: $1, body: $2) },
|
||||
logger: app.logger)
|
||||
app.lifecycle.use(BootTask { await digest.run() })
|
||||
}
|
||||
|
||||
// Prometheus, on its own listener. Off unless FN_METRICS_PORT is set, and
|
||||
// never the public port: one socket serving both would put the counters
|
||||
// on the internet.
|
||||
if config.metricsPort > 0 {
|
||||
guard config.metricsPort != config.port else {
|
||||
app.logger.critical("FN_METRICS_PORT must differ from FN_PORT")
|
||||
throw Config.ConfigError.missing("an FN_METRICS_PORT other than FN_PORT")
|
||||
}
|
||||
app.lifecycle.use(MetricsLifecycle(host: config.metricsHost, port: config.metricsPort))
|
||||
}
|
||||
|
||||
app.logger.info("relay listening on \(config.host):\(config.port) for \(config.audience)")
|
||||
}
|
||||
|
||||
|
||||
@@ -39,6 +39,7 @@ enum Federation {
|
||||
case unknownPeer(String)
|
||||
case badSignature
|
||||
case replayed
|
||||
case keyNotPinned(String)
|
||||
|
||||
var description: String {
|
||||
switch self {
|
||||
@@ -49,6 +50,8 @@ enum Federation {
|
||||
case .unknownPeer(let p): return "could not learn the signing key for \(p)"
|
||||
case .badSignature: return "federation signature does not verify"
|
||||
case .replayed: return "this federation request has already been seen"
|
||||
case .keyNotPinned(let p):
|
||||
return "the federation key \(p) publishes is not a key this relay pins for it"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -180,6 +183,8 @@ struct FederationVerifier: Sendable {
|
||||
let skew: Int
|
||||
let directory: PeerDirectory?
|
||||
let replay: any ReplayChecking
|
||||
/// Pinned keys per peer, base64. See `Config.federationPeerKeys`.
|
||||
var peerKeys: [String: [String]] = [:]
|
||||
|
||||
var enabled: Bool { policy != "closed" }
|
||||
|
||||
@@ -212,6 +217,13 @@ struct FederationVerifier: Sendable {
|
||||
let keyRaw = Data(base64Encoded: keyB64),
|
||||
let key = try? Curve25519.Signing.PublicKey(rawRepresentation: keyRaw)
|
||||
else { throw Federation.AuthError.unknownPeer(origin) }
|
||||
// A pinned peer is believed only with a pinned key. Two pins carry a
|
||||
// rotation: the old key until the peer switches, the new one after.
|
||||
if let pins = peerKeys[origin],
|
||||
!pins.contains(where: { Data(base64Encoded: $0) == keyRaw })
|
||||
{
|
||||
throw Federation.AuthError.keyNotPinned(origin)
|
||||
}
|
||||
|
||||
guard let sig = Data(base64Encoded: signature) else {
|
||||
throw Federation.AuthError.badSignature
|
||||
|
||||
@@ -0,0 +1,153 @@
|
||||
import Foundation
|
||||
import NIOConcurrencyHelpers
|
||||
import NIOCore
|
||||
import NIOHTTP1
|
||||
import NIOPosix
|
||||
import Vapor
|
||||
|
||||
/// Process-wide counters for the Prometheus endpoint.
|
||||
///
|
||||
/// Counts only, never ids or addresses. They start at zero with the process,
|
||||
/// which is what Prometheus expects of a counter.
|
||||
final class RelayMetrics: Sendable {
|
||||
static let shared = RelayMetrics()
|
||||
|
||||
enum Counter: String, CaseIterable, Sendable {
|
||||
case pushSent = "flamenet_push_sent_total"
|
||||
case pushFailed = "flamenet_push_failed_total"
|
||||
case limiterFailOpen = "flamenet_limiter_fail_open_total"
|
||||
case envelopesWritten = "flamenet_envelopes_written_total"
|
||||
case federationDepositsIn = "flamenet_federation_deposits_in_total"
|
||||
case federationDepositsOut = "flamenet_federation_deposits_out_total"
|
||||
|
||||
var help: String {
|
||||
switch self {
|
||||
case .pushSent: return "APNs pushes Apple accepted."
|
||||
case .pushFailed: return "APNs pushes refused, unreachable, or not signed."
|
||||
case .limiterFailOpen:
|
||||
return "Rate-limit checks allowed because Redis did not answer."
|
||||
case .envelopesWritten: return "Envelopes committed to the database."
|
||||
case .federationDepositsIn: return "Deposits accepted from peer relays."
|
||||
case .federationDepositsOut: return "Deposits a peer relay accepted from this one."
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private let counts = NIOLockedValueBox([Counter: Int]())
|
||||
|
||||
func add(_ counter: Counter, _ amount: Int = 1) {
|
||||
guard amount > 0 else { return }
|
||||
counts.withLockedValue { $0[counter, default: 0] += amount }
|
||||
}
|
||||
|
||||
func value(_ counter: Counter) -> Int {
|
||||
counts.withLockedValue { $0[counter] ?? 0 }
|
||||
}
|
||||
|
||||
/// The Prometheus text format, version 0.0.4.
|
||||
func render(heldPolls: Int, pushDropped: Int) -> String {
|
||||
var out = ""
|
||||
func metric(_ name: String, _ type: String, _ help: String, _ value: Int) {
|
||||
out += "# HELP \(name) \(help)\n# TYPE \(name) \(type)\n\(name) \(value)\n"
|
||||
}
|
||||
metric("flamenet_held_polls", "gauge", "Long-polls held open right now.", heldPolls)
|
||||
metric(
|
||||
"flamenet_push_dropped_total", "counter",
|
||||
"Pushes dropped because the push queue was full.", pushDropped)
|
||||
for counter in Counter.allCases {
|
||||
metric(counter.rawValue, "counter", counter.help, value(counter))
|
||||
}
|
||||
return out
|
||||
}
|
||||
}
|
||||
|
||||
/// A plain HTTP listener that answers `GET /metrics` and nothing else.
|
||||
///
|
||||
/// Its own socket on `FN_METRICS_PORT`, so the public port has no `/metrics`
|
||||
/// route at all.
|
||||
enum MetricsServer {
|
||||
static func start(
|
||||
host: String, port: Int, group: any EventLoopGroup,
|
||||
render: @escaping @Sendable () -> String
|
||||
) async throws -> any Channel {
|
||||
try await ServerBootstrap(group: group)
|
||||
.serverChannelOption(ChannelOptions.backlog, value: 16)
|
||||
.serverChannelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
|
||||
.childChannelInitializer { channel in
|
||||
channel.pipeline.configureHTTPServerPipeline().flatMap {
|
||||
channel.pipeline.addHandler(MetricsHandler(render: render))
|
||||
}
|
||||
}
|
||||
.bind(host: host, port: port)
|
||||
.get()
|
||||
}
|
||||
}
|
||||
|
||||
private final class MetricsHandler: ChannelInboundHandler, @unchecked Sendable {
|
||||
typealias InboundIn = HTTPServerRequestPart
|
||||
typealias OutboundOut = HTTPServerResponsePart
|
||||
|
||||
private let render: @Sendable () -> String
|
||||
private var head: HTTPRequestHead?
|
||||
|
||||
init(render: @escaping @Sendable () -> String) { self.render = render }
|
||||
|
||||
func channelRead(context: ChannelHandlerContext, data: NIOAny) {
|
||||
switch unwrapInboundIn(data) {
|
||||
case .head(let h): head = h
|
||||
case .body: break
|
||||
case .end:
|
||||
guard let head else { return }
|
||||
let path = head.uri.split(separator: "?", maxSplits: 1).first.map(String.init)
|
||||
let found = head.method == .GET && path == "/metrics"
|
||||
let body = found ? render() : "not found\n"
|
||||
var headers = HTTPHeaders()
|
||||
headers.add(
|
||||
name: "Content-Type",
|
||||
value: found ? "text/plain; version=0.0.4; charset=utf-8" : "text/plain")
|
||||
headers.add(name: "Content-Length", value: String(body.utf8.count))
|
||||
headers.add(name: "Connection", value: "close")
|
||||
let response = HTTPResponseHead(
|
||||
version: head.version, status: found ? .ok : .notFound, headers: headers)
|
||||
context.write(wrapOutboundOut(.head(response)), promise: nil)
|
||||
var buffer = context.channel.allocator.buffer(capacity: body.utf8.count)
|
||||
buffer.writeString(body)
|
||||
context.write(wrapOutboundOut(.body(.byteBuffer(buffer))), promise: nil)
|
||||
let channel = context.channel
|
||||
context.writeAndFlush(wrapOutboundOut(.end(nil))).whenComplete { _ in
|
||||
channel.close(promise: nil)
|
||||
}
|
||||
self.head = nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Starts the metrics listener after boot and closes it at shutdown.
|
||||
final class MetricsLifecycle: LifecycleHandler, @unchecked Sendable {
|
||||
private let host: String
|
||||
private let port: Int
|
||||
private var channel: (any Channel)?
|
||||
|
||||
init(host: String, port: Int) {
|
||||
self.host = host
|
||||
self.port = port
|
||||
}
|
||||
|
||||
func didBootAsync(_ application: Application) async throws {
|
||||
// Read once: the push worker is installed before boot, and a NIO
|
||||
// thread should not reach into application storage.
|
||||
let pushes = application.storage[PushWorker.Key.self]
|
||||
channel = try await MetricsServer.start(
|
||||
host: host, port: port, group: application.eventLoopGroup
|
||||
) {
|
||||
RelayMetrics.shared.render(
|
||||
heldPolls: RelayCapacity.shared.snapshot().heldNow,
|
||||
pushDropped: pushes?.dropped ?? 0)
|
||||
}
|
||||
application.logger.notice("metrics on http://\(host):\(port)/metrics")
|
||||
}
|
||||
|
||||
func shutdownAsync(_ application: Application) async {
|
||||
try? await channel?.close()
|
||||
}
|
||||
}
|
||||
@@ -867,6 +867,8 @@ enum RelayMigrations {
|
||||
CreateDeletedAccounts(),
|
||||
// Additive: two indexes, CONCURRENTLY on Postgres.
|
||||
AddScanIndexes(),
|
||||
// Additive: the report digest's watermark. Reports are untouched.
|
||||
CreateReportDigests(),
|
||||
]
|
||||
return list
|
||||
}
|
||||
|
||||
@@ -75,6 +75,7 @@ final class RedisRateLimiter: RateLimiting, @unchecked Sendable {
|
||||
limit: limit, now: now)
|
||||
}
|
||||
logger.error("rate limiter (redis) unavailable, allowing: \(error)")
|
||||
RelayMetrics.shared.add(.limiterFailOpen)
|
||||
return true
|
||||
}
|
||||
}
|
||||
@@ -103,6 +104,7 @@ final class RedisRateLimiter: RateLimiting, @unchecked Sendable {
|
||||
return await Self.localFallback.peek(key, limit: limit, now: now)
|
||||
}
|
||||
logger.error("rate limiter (redis) unavailable, allowing: \(error)")
|
||||
RelayMetrics.shared.add(.limiterFailOpen)
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
import Fluent
|
||||
import Foundation
|
||||
import Logging
|
||||
import SQLKit
|
||||
|
||||
/// One digest that went out: every report with `id <= upToID` has been mailed.
|
||||
///
|
||||
/// The watermark lives in the database so a restart neither loses nor repeats
|
||||
/// reports, and `up_to_id` is the primary key so two replicas that pick the
|
||||
/// same batch cannot both claim it.
|
||||
final class ReportDigestMark: Model, @unchecked Sendable {
|
||||
static let schema = "report_digests"
|
||||
|
||||
@ID(custom: "up_to_id", generatedBy: .user) var id: Int?
|
||||
@Field(key: "sent_at") var sentAt: Int
|
||||
|
||||
init() {}
|
||||
|
||||
init(upToID: Int, sentAt: Int) {
|
||||
self.id = upToID
|
||||
self.sentAt = sentAt
|
||||
}
|
||||
}
|
||||
|
||||
/// Seeded with the newest report already on disk, so the first digest after a
|
||||
/// deploy lists only reports filed after it. The ones before were mailed one
|
||||
/// at a time.
|
||||
struct CreateReportDigests: AsyncMigration {
|
||||
func prepare(on database: any Database) async throws {
|
||||
try await database.schema(ReportDigestMark.schema)
|
||||
.field("up_to_id", .int, .identifier(auto: false))
|
||||
.field("sent_at", .int, .required)
|
||||
.create()
|
||||
let newest = try await ContentReport.query(on: database).max(\.$id) ?? 0
|
||||
try await ReportDigestMark(upToID: newest, sentAt: 0).create(on: database)
|
||||
}
|
||||
|
||||
func revert(on database: any Database) async throws {
|
||||
try await database.schema(ReportDigestMark.schema).delete()
|
||||
}
|
||||
}
|
||||
|
||||
/// Mails the operator one message listing the reports filed since the last
|
||||
/// one, at most once an hour, and nothing when there are none.
|
||||
struct ReportDigest: Sendable {
|
||||
typealias Send =
|
||||
@Sendable (_ to: String, _ subject: String, _ body: String) async throws -> Void
|
||||
|
||||
let db: any Database
|
||||
let to: String
|
||||
let send: Send
|
||||
let logger: Logger
|
||||
|
||||
/// Shortest gap between two digests.
|
||||
static let interval = 3600
|
||||
/// How often the loop asks whether one is due.
|
||||
static let checkEvery: Duration = .seconds(300)
|
||||
/// Reports per message; the rest wait for the next one.
|
||||
static let maxPerDigest = 200
|
||||
/// Characters of each note and copy quoted in the message.
|
||||
static let excerpt = 1_000
|
||||
|
||||
func run() async {
|
||||
while !Task.isCancelled {
|
||||
do {
|
||||
_ = try await sendIfDue(now: Int(Date().timeIntervalSince1970))
|
||||
} catch {
|
||||
logger.error("report digest failed: \(error)")
|
||||
}
|
||||
try? await Task.sleep(for: Self.checkEvery)
|
||||
}
|
||||
}
|
||||
|
||||
/// Send one digest if an hour has passed since the last and there is
|
||||
/// something new. Returns how many reports it listed.
|
||||
@discardableResult
|
||||
func sendIfDue(now: Int) async throws -> Int {
|
||||
let last = try await ReportDigestMark.query(on: db).sort(\.$id, .descending).first()
|
||||
let watermark = last?.id ?? 0
|
||||
if let last, now - last.sentAt < Self.interval { return 0 }
|
||||
|
||||
let reports = try await ContentReport.query(on: db)
|
||||
.filter(\.$id > watermark)
|
||||
.sort(\.$id, .ascending)
|
||||
.limit(Self.maxPerDigest)
|
||||
.all()
|
||||
guard let newest = reports.last?.id else { return 0 }
|
||||
|
||||
// Claim the batch before mailing it. A second replica inserting the
|
||||
// same `up_to_id` fails on the primary key and sends nothing.
|
||||
let mark = ReportDigestMark(upToID: newest, sentAt: now)
|
||||
do {
|
||||
try await mark.create(on: db)
|
||||
} catch {
|
||||
logger.notice("report digest up to \(newest) already claimed")
|
||||
return 0
|
||||
}
|
||||
|
||||
do {
|
||||
try await send(
|
||||
to,
|
||||
"Flamenet reports: \(reports.count) new",
|
||||
Self.body(reports))
|
||||
} catch {
|
||||
// Released, so the next check retries the same reports. Loud,
|
||||
// because until then nobody has been told they are here.
|
||||
try? await mark.delete(on: db)
|
||||
logger.error(
|
||||
"report digest of \(reports.count) could not be mailed, retrying: \(error)")
|
||||
throw error
|
||||
}
|
||||
return reports.count
|
||||
}
|
||||
|
||||
static func body(_ reports: [ContentReport]) -> String {
|
||||
func clip(_ text: String) -> String {
|
||||
text.count > excerpt ? String(text.prefix(excerpt)) + " [cut]" : text
|
||||
}
|
||||
var lines = [
|
||||
"\(reports.count) report\(reports.count == 1 ? " was" : "s were") filed on this relay "
|
||||
+ "since the last digest.",
|
||||
"",
|
||||
]
|
||||
for r in reports {
|
||||
lines += [
|
||||
"id \(r.id.map(String.init) ?? "?")",
|
||||
"kind \(r.kind)",
|
||||
"reason \(r.reason)",
|
||||
"about user \(r.aboutID)",
|
||||
"filed by user \(r.reporterID)",
|
||||
"filed at \(ISO8601DateFormatter().string(from: Date(timeIntervalSince1970: TimeInterval(r.createdAt))))",
|
||||
"What they wrote:",
|
||||
r.note.isEmpty ? "(nothing)" : clip(r.note),
|
||||
"The copy they attached:",
|
||||
r.content.isEmpty ? "(none)" : clip(r.content),
|
||||
"",
|
||||
]
|
||||
}
|
||||
lines.append("Read the full text with Scripts/reports.sh 1 full.")
|
||||
return lines.joined(separator: "\n")
|
||||
}
|
||||
}
|
||||
@@ -72,6 +72,7 @@ extension Routes {
|
||||
let resolved = try await Self.resolveDeposits(input.envelopes, on: db)
|
||||
return (resolved.items, resolved.unknown)
|
||||
}
|
||||
RelayMetrics.shared.add(.federationDepositsIn)
|
||||
|
||||
return try .json(
|
||||
FederationDepositResponse(
|
||||
@@ -263,6 +264,7 @@ extension Routes {
|
||||
accepted += result.accepted
|
||||
unknown.append(contentsOf: result.unknown)
|
||||
reached.append(relay)
|
||||
RelayMetrics.shared.add(.federationDepositsOut)
|
||||
} catch FederationSender.SendError.notPermitted(let peer) {
|
||||
// Refusal is NOT unreachability.
|
||||
refused.append(peer)
|
||||
|
||||
@@ -68,6 +68,13 @@ extension Routes {
|
||||
guard submission.about != caller.userID else {
|
||||
throw WireError("bad_subject", "You cannot report yourself.", 400)
|
||||
}
|
||||
// Per reporter, per hour, so one account cannot bury the others.
|
||||
let admitted = await limits.consumeAll(
|
||||
[(key: "report:\(caller.userID)", amount: 1, limit: config.reportRatePerHour)],
|
||||
now: now)
|
||||
guard admitted else {
|
||||
throw WireError.rateLimited("Too many reports recently. Try again later.")
|
||||
}
|
||||
let content = String((submission.content ?? "").prefix(ContentReport.maximumContent))
|
||||
let note = String((submission.note ?? "").prefix(ContentReport.maximumContent))
|
||||
|
||||
@@ -81,46 +88,10 @@ extension Routes {
|
||||
report.createdAt = now
|
||||
try await report.save(on: request.db)
|
||||
|
||||
// Tell somebody. A row nobody is told about is a report nobody reads,
|
||||
// and the person who filed it has been told it reached the operator.
|
||||
//
|
||||
// Detached and never able to fail the report: if the mail cannot go
|
||||
// out, the row is still filed and `Scripts/reports.sh` still finds it.
|
||||
// Refusing the report because a mail server is down would lose the one
|
||||
// copy of something somebody went out of their way to send.
|
||||
if let to = reportsTo, let mailer = self.mailer {
|
||||
let id = report.id.map(String.init) ?? "unsaved"
|
||||
let summary = """
|
||||
A report was filed on this relay.
|
||||
|
||||
id \(id)
|
||||
kind \(report.kind)
|
||||
reason \(report.reason)
|
||||
about user \(report.aboutID)
|
||||
filed by user \(report.reporterID)
|
||||
|
||||
What they wrote:
|
||||
\(report.note.isEmpty ? "(nothing)" : report.note)
|
||||
|
||||
The copy they attached:
|
||||
\(report.content.isEmpty ? "(none)" : report.content)
|
||||
|
||||
Read the rest, and mark it seen, with Scripts/reports.sh.
|
||||
"""
|
||||
let logger = request.logger
|
||||
Task.detached {
|
||||
do {
|
||||
try await mailer.send(
|
||||
to: to,
|
||||
subject: "Flamenet report \(id): \(report.reason) (\(report.kind))",
|
||||
body: summary)
|
||||
} catch {
|
||||
// Loud, because the report is now only in the table and
|
||||
// nobody has been told it is there.
|
||||
logger.error("report \(id) filed but could not be mailed: \(error)")
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Told in the hourly digest (`ReportDigest`), one message for every
|
||||
// report since the last, read from this table. One mail per report let
|
||||
// one account fill the operator's inbox.
|
||||
if reportsTo == nil || mailer == nil {
|
||||
let id = report.id.map(String.init) ?? "?"
|
||||
request.logger.warning(
|
||||
"report \(id) filed with nowhere to send it; set FN_REPORTS_TO or nobody is told")
|
||||
|
||||
@@ -371,3 +371,96 @@ struct FederationPeerTests {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - Pinned federation keys
|
||||
|
||||
extension FederationPeerTests {
|
||||
|
||||
private func pinned(
|
||||
_ pins: [String], publishing key: Curve25519.Signing.PrivateKey
|
||||
) -> FederationVerifier {
|
||||
var (v, _) = verifier(descriptorBody: descriptor(audience: "peer.example", key: key))
|
||||
v.peerKeys = ["peer.example": pins]
|
||||
return v
|
||||
}
|
||||
|
||||
private func verifySigned(
|
||||
by key: Curve25519.Signing.PrivateKey, with v: FederationVerifier
|
||||
) async throws -> Federation.Peer {
|
||||
let s = signed(key)
|
||||
return try await v.verify(
|
||||
method: "POST", path: "/e2e/federation/deposit", body: Data(),
|
||||
origin: "peer.example", timestamp: s.timestamp, nonce: s.nonce,
|
||||
signature: s.signature, now: now)
|
||||
}
|
||||
|
||||
private func b64(_ key: Curve25519.Signing.PrivateKey) -> String {
|
||||
key.publicKey.rawRepresentation.base64EncodedString()
|
||||
}
|
||||
|
||||
@Test("a peer publishing its pinned key is accepted")
|
||||
func pinnedKeyAccepted() async throws {
|
||||
let key = Curve25519.Signing.PrivateKey()
|
||||
let peer = try await verifySigned(by: key, with: pinned([b64(key)], publishing: key))
|
||||
#expect(peer.audience == "peer.example")
|
||||
}
|
||||
|
||||
/// The descriptor and the signature agree with each other, so only the
|
||||
/// pin catches a peer whose descriptor was swapped.
|
||||
@Test("a peer publishing a key other than its pin is refused")
|
||||
func wrongKeyRefused() async throws {
|
||||
let key = Curve25519.Signing.PrivateKey()
|
||||
let other = Curve25519.Signing.PrivateKey()
|
||||
let v = pinned([b64(other)], publishing: key)
|
||||
var refusal: Federation.AuthError?
|
||||
do { _ = try await verifySigned(by: key, with: v) } catch let e as Federation.AuthError {
|
||||
refusal = e
|
||||
}
|
||||
guard case .keyNotPinned("peer.example") = refusal else {
|
||||
Issue.record("expected keyNotPinned, got \(String(describing: refusal))")
|
||||
return
|
||||
}
|
||||
#expect(refusal?.status == .unauthorized)
|
||||
}
|
||||
|
||||
@Test("during a rotation both pinned keys are accepted")
|
||||
func rotationAcceptsBoth() async throws {
|
||||
let old = Curve25519.Signing.PrivateKey()
|
||||
let new = Curve25519.Signing.PrivateKey()
|
||||
for key in [old, new] {
|
||||
let v = pinned([b64(old), b64(new)], publishing: key)
|
||||
let peer = try await verifySigned(by: key, with: v)
|
||||
#expect(peer.audience == "peer.example")
|
||||
}
|
||||
}
|
||||
|
||||
@Test("FN_FEDERATION_PEERS reads host=key1|key2 and keeps base64 padding")
|
||||
func peersParse() {
|
||||
let a = b64(Curve25519.Signing.PrivateKey())
|
||||
let b = b64(Curve25519.Signing.PrivateKey())
|
||||
#expect(a.hasSuffix("="))
|
||||
let parsed = Config.parseFederationPeers(
|
||||
" one.example , two.example=\(a)|\(b),three.example=\(a)")
|
||||
#expect(parsed.hosts == ["one.example", "two.example", "three.example"])
|
||||
#expect(parsed.keys["one.example"] == nil)
|
||||
#expect(parsed.keys["two.example"] == [a, b])
|
||||
#expect(parsed.keys["three.example"] == [a])
|
||||
}
|
||||
|
||||
@Test("a bare host boots with a warning that it is unpinned")
|
||||
func bareHostWarns() throws {
|
||||
let key = b64(Curve25519.Signing.PrivateKey())
|
||||
let config = try Config.fromEnvironment([
|
||||
"FN_ISSUER_PUBLIC_KEY": key,
|
||||
"FN_RELAY_AUDIENCE": "us.example",
|
||||
"FN_FEDERATION": "allowlist",
|
||||
"FN_FEDERATION_PEERS": "bare.example,pinned.example=\(key),bad.example=nope",
|
||||
])
|
||||
#expect(config.federationPeers == ["bare.example", "pinned.example", "bad.example"])
|
||||
let warnings = config.federationPinWarnings()
|
||||
#expect(warnings.count == 2)
|
||||
#expect(warnings.contains { $0.contains("bare.example is unpinned") })
|
||||
#expect(warnings.contains { $0.contains("bad.example has a pinned key that is not") })
|
||||
#expect(!warnings.contains { $0.contains("pinned.example") })
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
import Crypto
|
||||
import Foundation
|
||||
import NIOCore
|
||||
import NIOPosix
|
||||
import Testing
|
||||
import Vapor
|
||||
import VaporTesting
|
||||
|
||||
@testable import FlamenetRelay
|
||||
|
||||
/// The Prometheus endpoint: on its own port, never the public one.
|
||||
@Suite("metrics")
|
||||
struct MetricsTests {
|
||||
|
||||
/// A port nothing is listening on, found by binding 0 and letting go.
|
||||
private func freePort() async throws -> Int {
|
||||
let channel = try await ServerBootstrap(group: MultiThreadedEventLoopGroup.singleton)
|
||||
.bind(host: "127.0.0.1", port: 0).get()
|
||||
let port = channel.localAddress?.port ?? 0
|
||||
try await channel.close()
|
||||
return port
|
||||
}
|
||||
|
||||
private func withRelayApp(
|
||||
metricsPort: Int, _ body: (Application) async throws -> Void
|
||||
) async throws {
|
||||
try await ProcessEnvironmentLock.withExclusiveEnvironment {
|
||||
let issuer = Curve25519.Signing.PrivateKey()
|
||||
let dbPath = NSTemporaryDirectory() + "fn-metrics-\(UUID().uuidString).sqlite"
|
||||
let attachments = NSTemporaryDirectory() + "fn-metrics-att-\(UUID().uuidString)"
|
||||
let env: [String: String] = [
|
||||
"FN_ISSUER_PUBLIC_KEY": issuer.publicKey.rawRepresentation.base64EncodedString(),
|
||||
"FN_RELAY_AUDIENCE": "test.relay",
|
||||
"FN_DB": dbPath,
|
||||
"FN_ATTACHMENTS_DIR": attachments,
|
||||
"FN_LOG_LEVEL": "critical",
|
||||
"FN_METRICS_PORT": String(metricsPort),
|
||||
]
|
||||
let previous = env.keys.reduce(into: [String: String?]()) {
|
||||
$0[$1] = ProcessInfo.processInfo.environment[$1]
|
||||
}
|
||||
for (k, v) in env { setenv(k, v, 1) }
|
||||
defer {
|
||||
for (k, v) in previous { if let v { setenv(k, v, 1) } else { unsetenv(k) } }
|
||||
try? FileManager.default.removeItem(atPath: dbPath)
|
||||
try? FileManager.default.removeItem(atPath: attachments)
|
||||
}
|
||||
let app = try await Application.make(.testing)
|
||||
do {
|
||||
try await configure(app)
|
||||
// What `serve` does; the metrics listener starts here.
|
||||
try await app.asyncBoot()
|
||||
try await body(app)
|
||||
} catch {
|
||||
try? await app.asyncShutdown()
|
||||
throw error
|
||||
}
|
||||
try? await app.asyncShutdown()
|
||||
}
|
||||
}
|
||||
|
||||
@Test("the counters render in the Prometheus text format")
|
||||
func renders() {
|
||||
RelayMetrics.shared.add(.envelopesWritten, 2)
|
||||
RelayMetrics.shared.add(.federationDepositsIn)
|
||||
let text = RelayMetrics.shared.render(heldPolls: 7, pushDropped: 3)
|
||||
#expect(text.contains("# TYPE flamenet_held_polls gauge\nflamenet_held_polls 7\n"))
|
||||
#expect(text.contains("flamenet_push_dropped_total 3\n"))
|
||||
for name in [
|
||||
"flamenet_push_sent_total", "flamenet_push_failed_total",
|
||||
"flamenet_limiter_fail_open_total", "flamenet_envelopes_written_total",
|
||||
"flamenet_federation_deposits_in_total", "flamenet_federation_deposits_out_total",
|
||||
] {
|
||||
#expect(text.contains("# TYPE \(name) counter\n\(name) "), "missing \(name)")
|
||||
}
|
||||
#expect(RelayMetrics.shared.value(.envelopesWritten) >= 2)
|
||||
}
|
||||
|
||||
@Test("/metrics is absent on the public port and served on the metrics port")
|
||||
func separatePort() async throws {
|
||||
let port = try await freePort()
|
||||
try await withRelayApp(metricsPort: port) { app in
|
||||
try await app.testing().test(.GET, "/metrics") { res in
|
||||
#expect(res.status == .notFound)
|
||||
#expect(!res.body.string.contains("flamenet_held_polls"))
|
||||
}
|
||||
|
||||
let res = try await app.client.get(URI(string: "http://127.0.0.1:\(port)/metrics"))
|
||||
#expect(res.status == .ok)
|
||||
#expect(res.headers.first(name: .contentType)?.hasPrefix("text/plain") == true)
|
||||
let text = res.body.map { String(buffer: $0) } ?? ""
|
||||
for name in [
|
||||
"flamenet_held_polls", "flamenet_push_sent_total", "flamenet_push_failed_total",
|
||||
"flamenet_push_dropped_total", "flamenet_limiter_fail_open_total",
|
||||
"flamenet_envelopes_written_total", "flamenet_federation_deposits_in_total",
|
||||
] {
|
||||
#expect(text.contains("\n\(name) "), "missing \(name)")
|
||||
}
|
||||
|
||||
// Only /metrics, on that port too.
|
||||
let other = try await app.client.get(URI(string: "http://127.0.0.1:\(port)/health"))
|
||||
#expect(other.status == .notFound)
|
||||
}
|
||||
}
|
||||
|
||||
@Test("metrics are off unless FN_METRICS_PORT is set, and bind loopback by default")
|
||||
func offByDefault() throws {
|
||||
func config(_ extra: [String: String]) throws -> Config {
|
||||
var env = [
|
||||
"FN_ISSUER_PUBLIC_KEY": Curve25519.Signing.PrivateKey()
|
||||
.publicKey.rawRepresentation.base64EncodedString(),
|
||||
"FN_RELAY_AUDIENCE": "test.relay",
|
||||
]
|
||||
for (k, v) in extra { env[k] = v }
|
||||
return try Config.fromEnvironment(env)
|
||||
}
|
||||
#expect(try config([:]).metricsPort == 0)
|
||||
#expect(try config([:]).metricsHost == "127.0.0.1")
|
||||
#expect(try config(["FN_METRICS_PORT": "9464"]).metricsPort == 9464)
|
||||
}
|
||||
|
||||
@Test("a metrics port equal to the public port refuses to boot")
|
||||
func samePortRefused() async throws {
|
||||
await #expect(throws: (any Error).self) {
|
||||
try await withRelayApp(metricsPort: 8080) { _ in }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
import Crypto
|
||||
import Fluent
|
||||
import Foundation
|
||||
import NIOConcurrencyHelpers
|
||||
import Testing
|
||||
import Vapor
|
||||
|
||||
@testable import FlamenetRelay
|
||||
|
||||
/// The per-reporter limit, and the hourly digest that replaced one mail per
|
||||
/// report.
|
||||
@Suite("report digest")
|
||||
struct ReportDigestTests {
|
||||
|
||||
private let now = 1_700_000_000
|
||||
|
||||
/// Every message the digest would have sent.
|
||||
private final class Outbox: @unchecked Sendable {
|
||||
private let box = NIOLockedValueBox<[(to: String, subject: String, body: String)]>([])
|
||||
var failNext = false
|
||||
var sent: [(to: String, subject: String, body: String)] { box.withLockedValue { $0 } }
|
||||
|
||||
func send(_ to: String, _ subject: String, _ body: String) throws {
|
||||
if failNext {
|
||||
failNext = false
|
||||
throw Abort(.serviceUnavailable)
|
||||
}
|
||||
box.withLockedValue { $0.append((to, subject, body)) }
|
||||
}
|
||||
}
|
||||
|
||||
private func digest(_ h: RelayHarness, _ outbox: Outbox) -> ReportDigest {
|
||||
ReportDigest(
|
||||
db: h.db, to: "ops@example.org",
|
||||
send: { try outbox.send($0, $1, $2) },
|
||||
logger: h.app.logger)
|
||||
}
|
||||
|
||||
@discardableResult
|
||||
private func fileRow(_ h: RelayHarness, about: Int, note: String) async throws -> Int {
|
||||
let r = ContentReport()
|
||||
r.reporterID = 1
|
||||
r.aboutID = about
|
||||
r.kind = "message"
|
||||
r.reason = "spam"
|
||||
r.content = "copy \(about)"
|
||||
r.note = note
|
||||
r.createdAt = now
|
||||
try await r.save(on: h.db)
|
||||
return r.id!
|
||||
}
|
||||
|
||||
@Test("one reporter over the hourly limit gets a 429")
|
||||
func reportLimit() async throws {
|
||||
try await withRelay(env: ["FN_REPORT_RATE": "2"]) { h in
|
||||
let d = h.device(user: 1, id: "aaaa-0001")
|
||||
_ = try await h.register(d, on: h.db)
|
||||
func file() async throws -> Response {
|
||||
let body = Data(#"{"about":2,"kind":"story","reason":"spam"}"#.utf8)
|
||||
let req = h.request(.POST, "/e2e/moderation/report", as: d, body: body, now: now)
|
||||
return try await h.routes.submitReport(req, now: now)
|
||||
}
|
||||
#expect(try await file().status == .ok)
|
||||
#expect(try await file().status == .ok)
|
||||
var status = 0
|
||||
do { _ = try await file() } catch let e as WireError { status = e.data.status }
|
||||
#expect(status == 429)
|
||||
// The refused one was not stored.
|
||||
#expect(try await ContentReport.query(on: h.db).count() == 2)
|
||||
}
|
||||
}
|
||||
|
||||
@Test("the default limit is 10 an hour")
|
||||
func reportLimitDefault() throws {
|
||||
let config = try Config.fromEnvironment([
|
||||
"FN_ISSUER_PUBLIC_KEY": Curve25519.Signing.PrivateKey()
|
||||
.publicKey.rawRepresentation.base64EncodedString(),
|
||||
"FN_RELAY_AUDIENCE": "test.relay",
|
||||
])
|
||||
#expect(config.reportRatePerHour == 10)
|
||||
}
|
||||
|
||||
@Test("N reports go out as one email, and an empty hour sends none")
|
||||
func batchesIntoOne() async throws {
|
||||
try await withRelay { h in
|
||||
let outbox = Outbox()
|
||||
let digest = digest(h, outbox)
|
||||
|
||||
// Nothing filed: nothing sent.
|
||||
#expect(try await digest.sendIfDue(now: now) == 0)
|
||||
#expect(outbox.sent.isEmpty)
|
||||
|
||||
var ids: [Int] = []
|
||||
for about in 2...4 { ids.append(try await fileRow(h, about: about, note: "n\(about)")) }
|
||||
#expect(try await digest.sendIfDue(now: now) == 3)
|
||||
#expect(outbox.sent.count == 1)
|
||||
let mail = try #require(outbox.sent.first)
|
||||
#expect(mail.to == "ops@example.org")
|
||||
#expect(mail.subject == "Flamenet reports: 3 new")
|
||||
for (id, about) in zip(ids, 2...4) {
|
||||
#expect(mail.body.contains("id \(id)"))
|
||||
#expect(mail.body.contains("about user \(about)"))
|
||||
#expect(mail.body.contains("n\(about)"))
|
||||
}
|
||||
|
||||
// The same reports are not mailed twice, and an hour with nothing
|
||||
// new sends nothing.
|
||||
#expect(try await digest.sendIfDue(now: now + 4000) == 0)
|
||||
#expect(outbox.sent.count == 1)
|
||||
|
||||
// A report inside the hour waits for the hour.
|
||||
try await fileRow(h, about: 5, note: "later")
|
||||
#expect(try await digest.sendIfDue(now: now + 600) == 0)
|
||||
#expect(outbox.sent.count == 1)
|
||||
#expect(try await digest.sendIfDue(now: now + 3600) == 1)
|
||||
#expect(outbox.sent.count == 2)
|
||||
#expect(outbox.sent[1].body.contains("later"))
|
||||
#expect(!outbox.sent[1].body.contains("n2"))
|
||||
|
||||
// The stored reports are untouched.
|
||||
#expect(try await ContentReport.query(on: h.db).count() == 4)
|
||||
}
|
||||
}
|
||||
|
||||
@Test("a digest that could not be mailed is retried with the same reports")
|
||||
func failedSendRetries() async throws {
|
||||
try await withRelay { h in
|
||||
let outbox = Outbox()
|
||||
let digest = digest(h, outbox)
|
||||
try await fileRow(h, about: 2, note: "first")
|
||||
outbox.failNext = true
|
||||
await #expect(throws: (any Error).self) { try await digest.sendIfDue(now: now) }
|
||||
#expect(outbox.sent.isEmpty)
|
||||
#expect(try await digest.sendIfDue(now: now + 300) == 1)
|
||||
#expect(outbox.sent.first?.body.contains("first") == true)
|
||||
}
|
||||
}
|
||||
|
||||
/// The migration seeds the watermark with the newest report, so a deploy
|
||||
/// does not mail every report ever filed again.
|
||||
@Test("reports filed before the digest existed are not mailed again")
|
||||
func seededAtTheNewestReport() async throws {
|
||||
try await withRelay { h in
|
||||
try await fileRow(h, about: 2, note: "old")
|
||||
// What the migration does on a database that already has reports.
|
||||
try await CreateReportDigests().revert(on: h.db)
|
||||
try await CreateReportDigests().prepare(on: h.db)
|
||||
let outbox = Outbox()
|
||||
#expect(try await digest(h, outbox).sendIfDue(now: now) == 0)
|
||||
#expect(outbox.sent.isEmpty)
|
||||
}
|
||||
}
|
||||
}
|
||||
+54
-4
@@ -51,10 +51,14 @@ running one box either way.
|
||||
|
||||
## Storage
|
||||
|
||||
**A self-hosted relay needs Postgres.** SQLite is for local development only
|
||||
and receives no new features. Delete-on-ack already runs only on Postgres,
|
||||
because SQLite reuses row ids.
|
||||
|
||||
| Variable | Default | Notes |
|
||||
|---|---|---|
|
||||
| `FN_POSTGRES_URL` | unset | Set it and Postgres is used. |
|
||||
| `FN_DB` | `relay.sqlite` | SQLite file, used when the above is unset. **The schema is identical on both**: SQLite is not a degraded mode, and is the right choice for a relay serving a household or a company. |
|
||||
| `FN_POSTGRES_URL` | unset | Set it and Postgres is used. Required for any relay with real accounts. |
|
||||
| `FN_DB` | `relay.sqlite` | SQLite file, used when the above is unset. Local development only. |
|
||||
| `FN_ATTACHMENTS_DIR` | `attachments` | Blobs are files, not rows, in both modes. |
|
||||
| `FN_WORKDIR` | the process's own directory | Where every relative path above is resolved from. |
|
||||
| `FN_S3_ENDPOINT` | unset | Set it and attachments go to object storage instead of `FN_ATTACHMENTS_DIR`. |
|
||||
@@ -103,9 +107,28 @@ should stand alone without configuring anything.
|
||||
| Variable | Default | Notes |
|
||||
|---|---|---|
|
||||
| `FN_FEDERATION` | `closed` | `closed`, `allowlist`, or `open`. An unrecognised value falls back to `closed` rather than refusing to boot, a typo in an optional variable should not stop a relay serving its own users. |
|
||||
| `FN_FEDERATION_PEERS` | empty | Comma-separated hostnames, used only under `allowlist`. Enforced **outbound as well as inbound**, from the same rule. Peers named here are fetched wherever they resolve: the outbound address guard, which refuses private and loopback addresses, is for names an attacker can choose under `open` federation, not for a name you typed and pinned by key. Three relays on one host resolve each other to private addresses, and would otherwise refuse every deposit between them as unreachable. |
|
||||
| `FN_FEDERATION_PEERS` | empty | Comma-separated peers, each `host`, `host=key` or `host=key1\|key2`. The key is the peer's federation public key, base64, as its `/.well-known/flamenet/relay` publishes it in `federationKey`. A pinned peer whose descriptor publishes any other key is refused with 401. A bare `host` accepts whatever key the descriptor publishes, and boot logs a warning naming it. The host list is used only under `allowlist`; pins apply under `allowlist` and `open`. Enforced **outbound as well as inbound**, from the same rule. Peers named here are fetched wherever they resolve: the outbound address guard, which refuses private and loopback addresses, is for names an attacker can choose under `open` federation, not for a name you typed and pinned by key. Three relays on one host resolve each other to private addresses, and would otherwise refuse every deposit between them as unreachable. |
|
||||
| `FN_FEDERATION_KEY` | unset | Base64 Ed25519 seed. **Required** when `FN_FEDERATION` is not `closed`: the relay refuses to boot otherwise, rather than starting unreachable and letting a stranger discover it. |
|
||||
|
||||
### Pinning and rotating a peer's key
|
||||
|
||||
Read the peer's key and pin it:
|
||||
|
||||
```bash
|
||||
curl -s https://peer.example/.well-known/flamenet/relay | jq -r .federationKey
|
||||
FN_FEDERATION_PEERS=peer.example=<that key>
|
||||
```
|
||||
|
||||
A peer rotating its key tells you the new public key first. Then:
|
||||
|
||||
1. Pin both, old first: `peer.example=<old>|<new>`. Restart. Both are accepted.
|
||||
2. The peer switches `FN_FEDERATION_KEY` to the new seed and restarts.
|
||||
3. Wait an hour: the relay caches a peer's descriptor for 3600 seconds.
|
||||
4. Drop the old key: `peer.example=<new>`. Restart.
|
||||
|
||||
Rotating your own key is the same steps from the other side: send each peer
|
||||
the new public key, wait until all of them pin both, then switch.
|
||||
|
||||
`FN_FEDERATION_KEY` is deliberately **not** the issuer key. A key that only ever
|
||||
needs to say "this relay said this" should not also be able to mint a token for
|
||||
any user on it.
|
||||
@@ -201,7 +224,8 @@ On. A person can report something to whoever runs the server, so blocking is not
|
||||
| Variable | Default | What it does |
|
||||
|---|---|---|
|
||||
| `FN_CONTENT_REPORTS` | on | Set to `false`, `0`, `no` or `off` to stop taking reports. Anything else, including a typo, leaves them on. |
|
||||
| `FN_REPORTS_TO` | unset | Where a filed report is mailed, through the SMTP settings. Unset, reports are kept in `content_reports` and read with `Scripts/reports.sh`, and the log says nobody was told. |
|
||||
| `FN_REPORTS_TO` | unset | Where the report digest is mailed, through the SMTP settings. One message lists every report filed since the last one, at most one an hour, checked every 5 minutes; an hour with no reports sends nothing. At most 200 reports per message, each note and copy cut at 1,000 characters. A digest that fails to send is retried with the same reports. The watermark is the `report_digests` table, so a restart neither loses nor repeats a report. Unset, reports are kept in `content_reports` and read with `Scripts/reports.sh`, and the log says nobody was told. |
|
||||
| `FN_REPORT_RATE` | `10`/hr | Reports one account may file per hour. Over it, the route answers 429 and nothing is stored. |
|
||||
| `FN_REPORT_RETENTION_DAYS` | `365` | How long a content report, and the plaintext copy the reporter chose to attach, is kept. Long enough to show a pattern, bounded so it is not forever. Shown on /metadata. |
|
||||
|
||||
Off is a legitimate answer, not a broken relay. If nobody is going to read them, collecting reports is a promise the server cannot keep, and saying so plainly is better than a queue nobody opens.
|
||||
@@ -327,6 +351,32 @@ service (`--all` adds retired ones) and `Scripts/bridges.sh retire <fingerprint>
|
||||
takes one out of service. A retired identity is never listed again. The script
|
||||
header shows how to list a bridge on another machine.
|
||||
|
||||
## Metrics
|
||||
|
||||
Off. Set `FN_METRICS_PORT` and the relay opens a second listener that answers
|
||||
`GET /metrics` in the Prometheus text format and nothing else. The public port
|
||||
never serves `/metrics`, and the relay refuses to boot when the two ports are
|
||||
the same.
|
||||
|
||||
| Variable | Default | Notes |
|
||||
|---|---|---|
|
||||
| `FN_METRICS_PORT` | unset (off) | Port for the metrics listener. |
|
||||
| `FN_METRICS_HOST` | `127.0.0.1` | Bind address. In Docker, set `0.0.0.0` so a scraper on the same network can reach it, and do not publish the port. |
|
||||
|
||||
| Metric | Type | Counts |
|
||||
|---|---|---|
|
||||
| `flamenet_held_polls` | gauge | Long-polls held open now. |
|
||||
| `flamenet_push_sent_total` | counter | APNs pushes Apple accepted. |
|
||||
| `flamenet_push_failed_total` | counter | APNs pushes refused, unreachable or not signed. |
|
||||
| `flamenet_push_dropped_total` | counter | Pushes dropped because the push queue was full. |
|
||||
| `flamenet_limiter_fail_open_total` | counter | Rate-limit checks allowed because Redis did not answer. |
|
||||
| `flamenet_envelopes_written_total` | counter | Envelopes committed. |
|
||||
| `flamenet_federation_deposits_in_total` | counter | Deposits accepted from peers. |
|
||||
| `flamenet_federation_deposits_out_total` | counter | Deposits a peer accepted from this relay. |
|
||||
|
||||
Counters start at zero with the process. No metric carries an id, a name or an
|
||||
address.
|
||||
|
||||
## Site copy
|
||||
|
||||
| Variable | Default | Notes |
|
||||
|
||||
Reference in New Issue
Block a user