1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496

// observations-inbox accepts audio replies to observations episodes.
//
// It is the endpoint behind the drop zone on observations.profpatsch.de: a
// listener drops a recording, it is stored in one SQLite file, and a
// notification with an unguessable link arrives by mail. There is no listing
// route, no account, and no way to get from one submission to another.
//
// See observations-inbox(1) for the reference and observations(7) for why the
// format has an inbox at all.
package main

import (
	"context"
	"database/sql"
	"errors"
	"flag"
	"fmt"
	"log"
	"net/http"
	"os"
	"strings"
	"time"
)

type server struct {
	db         *sql.DB
	baseURL    string
	passphrase string
	limiter    *rateLimiter
	mail       *mailer
	geminiKey  string
	ffprobeBin string
	ffmpegBin  string

	// quotaWarned keeps the "inbox is full" mail to one per filling, so a
	// scanner hammering a full inbox cannot turn it into a mail flood.
	quotaWarned bool
}

func main() {
	log.SetFlags(0)
	log.SetPrefix("observations-inbox: ")

	var (
		addr           = flag.String("addr", "127.0.0.1:8778", "address to listen on")
		dbPath         = flag.String("db", "/var/lib/observations/inbox.db", "path to the SQLite database")
		baseURL        = flag.String("base-url", "https://observations.profpatsch.de", "public base URL, used to build submission links")
		passphraseFile = flag.String("passphrase-file", "", "file holding the submission passphrase")
		mailFile       = flag.String("mail-file", "", "file holding the mail configuration (empty disables mail)")
		geminiKeyFile  = flag.String("gemini-key-file", "", "file holding the Gemini API key (empty disables transcription)")
		ffprobeBin     = flag.String("ffprobe", "ffprobe", "path to ffprobe, used to measure how long a recording is")
		ffmpegBin      = flag.String("ffmpeg", "ffmpeg", "path to ffmpeg, used to give a recording a duration its player can see")
	)
	flag.Parse()

	if *passphraseFile == "" {
		log.Fatal("--passphrase-file is required: without it anyone could submit")
	}
	passphrase, err := readSecret(*passphraseFile)
	if err != nil {
		log.Fatalf("reading passphrase: %v", err)
	}
	if normalisePassphrase(passphrase) == "" {
		log.Fatalf("passphrase in %s has no letters or digits", *passphraseFile)
	}

	db, err := openDB(*dbPath)
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	s := &server{
		db:         db,
		baseURL:    strings.TrimRight(*baseURL, "/"),
		passphrase: passphrase,
		limiter:    newRateLimiter(submitRate, submitRateWind),
		ffprobeBin: *ffprobeBin,
		ffmpegBin:  *ffmpegBin,
	}

	if *mailFile != "" {
		m, err := loadMailer(*mailFile)
		if err != nil {
			log.Fatalf("reading mail configuration: %v", err)
		}
		s.mail = m
		// The host and address are not secret in the sense the password is,
		// but they are the account this thing sends as, so the log says only
		// that mail is configured.
		log.Print("notifications enabled")
	} else {
		log.Print("no --mail-file: submissions will be stored but not announced")
	}

	if *geminiKeyFile != "" {
		if s.geminiKey, err = readSecret(*geminiKeyFile); err != nil {
			log.Fatalf("reading Gemini key: %v", err)
		}
		log.Print("transcription available (only used when the sender consents)")
	}

	// Recordings that arrived before anything measured them are measured now,
	// in the background: a submission already in the inbox is exactly as
	// unplayable as a new one would be, and the fix belongs to both.
	go s.measureBacklog()

	mux := http.NewServeMux()
	mux.HandleFunc("POST /submit", s.handleSubmit)
	mux.HandleFunc("GET /submit", s.handleSubmitStatus)
	mux.HandleFunc("GET /inbox/{token}", s.handleReview)
	mux.HandleFunc("GET /inbox/{token}/", s.handleReview)
	mux.HandleFunc("GET /inbox/{token}/audio", s.handleReviewAudio)
	mux.HandleFunc("POST /inbox/{token}/delete", s.handleReviewDelete)

	srv := &http.Server{
		Addr:    *addr,
		Handler: mux,
		// A submission is up to 5 MB over a phone connection, so the write
		// and read timeouts have to tolerate a slow uplink.
		ReadTimeout:  5 * time.Minute,
		WriteTimeout: 5 * time.Minute,
		IdleTimeout:  2 * time.Minute,
	}
	log.Printf("listening on %s", *addr)
	if err := srv.ListenAndServe(); err != nil {
		log.Fatal(err)
	}
}

// handleSubmitStatus reports whether the inbox is open, so the drop zone can
// say so before someone records a reply and finds out the hard way.
func (s *server) handleSubmitStatus(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Cache-Control", "no-store")
	w.Header().Set("Content-Type", "application/json")
	used, err := usedBytes(s.db)
	if err != nil {
		s.fail(w, err)
		return
	}
	open := used < quotaBytes
	fmt.Fprintf(w, `{"open":%t,"maxBytes":%d}`+"\n", open, int64(maxUploadBytes))
}

// handleSubmit accepts one recording.
//
// The order of checks is deliberate: everything that can be decided without
// touching the disk or the network happens first, so that abuse costs as
// little as possible.
func (s *server) handleSubmit(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Cache-Control", "no-store")

	ip := clientIP(r)

	filename, data, err := readUpload(w, r)
	if err != nil {
		s.refuse(w, http.StatusBadRequest, err.Error())
		return
	}

	// The passphrase is checked after reading the body because the body has
	// to be consumed either way for the connection to be reusable, but before
	// anything is stored or sent anywhere.
	if !passphraseOK(s.passphrase, r.Form.Get("passphrase")) {
		s.refuse(w, http.StatusForbidden,
			"That passphrase is not right. It is said at the start of each episode.")
		return
	}

	kind, err := sniffAudio(data)
	if err != nil {
		s.refuse(w, http.StatusUnsupportedMediaType,
			"That does not look like an audio recording.")
		return
	}

	// Rate limiting is counted here, once a submission is known to be a
	// genuine attempt: a valid passphrase and actual audio. Counting earlier
	// would let a stream of junk from one address use up the budget of
	// whoever is behind the same NAT, which is the opposite of the intent —
	// the limit exists to stop repeated *successful* submissions, and junk is
	// already refused on its own merits.
	if !s.limiter.allow(ip) {
		s.refuse(w, http.StatusTooManyRequests,
			"That is a lot of submissions in a short time. Try again later.")
		return
	}

	used, err := usedBytes(s.db)
	if err != nil {
		s.fail(w, err)
		return
	}
	if err := checkQuota(used, int64(len(data))); err != nil {
		s.notifyQuotaFull(used)
		s.refuse(w, http.StatusInsufficientStorage,
			"The inbox is full at the moment. Please try again in a few days.")
		return
	}

	consented := isTruthy(r.Form.Get("consent"))
	sub := &Submission{
		MIME:      kind.MIME,
		Filename:  sanitiseFilename(filename),
		Consented: consented,
		Note:      strings.TrimSpace(truncate(r.Form.Get("note"), 2000)),
		SourceIP:  ip,
	}
	if err := insertSubmission(s.db, sub, data); err != nil {
		s.fail(w, err)
		return
	}

	// Stored and safe. Everything after this point is best effort: the
	// submitter is told it worked, and any failure to transcribe or notify is
	// our problem, not theirs.
	w.Header().Set("Content-Type", "application/json")
	fmt.Fprint(w, `{"ok":true}`+"\n")

	go s.afterSubmit(sub, data, kind, used+int64(len(data)))
}

// afterSubmit measures, transcribes (if allowed) and notifies, after the
// response has gone out.
//
// None of this is on the submitter's critical path. Probing decodes or rewrites
// a container and transcription is a network round trip to Google; making
// someone on a phone connection wait for either, after their upload has already
// arrived intact, would be waiting for our benefit rather than theirs.
func (s *server) afterSubmit(sub *Submission, audio []byte, kind *audioType, used int64) {
	defer func() {
		if r := recover(); r != nil {
			log.Printf("panic while handling submission: %v", r)
		}
	}()

	mime := kind.MIME

	// Measure first, so the notification can say how long the recording is.
	// A failure here costs a number on a page, so it is logged and dropped.
	if res, err := probeAudio(context.Background(), s.ffprobeBin, s.ffmpegBin, audio, kind.Ext); err != nil {
		log.Printf("probing submission: %v", err)
	} else if res.DurationMs > 0 {
		sub.DurationMs = sql.NullInt64{Int64: res.DurationMs, Valid: true}
		switch {
		case res.Audio != nil:
			// The container was rewritten so that its length is visible
			// without decoding the whole file. Same audio, new box.
			if err := setRemuxedAudio(s.db, sub.Token, res.Audio, res.DurationMs); err != nil {
				log.Printf("storing remuxed audio: %v", err)
			} else {
				sub.ByteSize = int64(len(res.Audio))
				used = used - int64(len(audio)) + int64(len(res.Audio))
			}
		default:
			if err := setDuration(s.db, sub.Token, res.DurationMs); err != nil {
				log.Printf("storing duration: %v", err)
			}
		}
	}

	// Why there is no transcript is recorded with the submission, not only
	// logged and mailed. The mail is sent once and the log rotates, but the
	// review page is looked at whenever the link is clicked — and a page that
	// simply omits the transcript cannot say whether transcription failed,
	// was never configured, or was never permitted. Recording the reason is
	// what makes those distinguishable. It is not retried; see
	// observations-inbox(1).
	var transcriptErr error
	switch {
	case !sub.Consented:
		// Nothing was attempted and nothing left the machine. The consented
		// column already says this, so there is no reason to record.

	case s.geminiKey == "":
		// Consent was given and then nothing happened, which is the most
		// confusing silence of the three: it looks like a failure and is
		// really a server without a key.
		transcriptErr = errors.New("transcription is not configured on this server")
		log.Printf("submission consented to transcription, but no Gemini key is configured")
		s.recordTranscriptError(sub.Token, transcriptErr)

	default:
		text, err := transcribe(context.Background(), s.geminiKey, audio, mime)
		switch {
		case err != nil:
			transcriptErr = err
			log.Printf("transcribing submission: %v", err)
			s.recordTranscriptError(sub.Token, err)
		default:
			if err := setTranscript(s.db, sub.Token, text); err != nil {
				log.Printf("storing transcript: %v", err)
			}
			sub.Transcript = sql.NullString{String: text, Valid: true}
		}
	}

	if s.mail.enabled() {
		subject, text, html := submissionMail(s.baseURL, sub, used, transcriptErr)
		if err := s.mail.notify(subject, text, html); err != nil {
			// The submission is stored, so this is recoverable by hand: the
			// log carries the link.
			log.Printf("sending notification: %v (submission at %s/inbox/%s)",
				err, s.baseURL, sub.Token)
		}
	} else {
		log.Printf("submission stored: %s/inbox/%s", s.baseURL, sub.Token)
	}
}

// recordTranscriptError stores why a submission has no transcript.
//
// Best effort, like everything else after the response has gone out: failing
// to record why a transcript is missing must never be louder than the missing
// transcript itself, and the recording is untouched either way.
func (s *server) recordTranscriptError(token string, cause error) {
	if err := setTranscriptError(s.db, token, cause.Error()); err != nil {
		log.Printf("storing transcript error: %v", err)
	}
}

// measureBacklog gives a length to submissions stored before there was
// anything to measure them.
//
// It runs once at startup, after the listener is up, and never blocks it: the
// service is useful without this, and a recording that cannot be measured is
// only missing a number. Nothing here is notified — these submissions were
// announced when they arrived, and a second mail about a recording already
// dealt with would be noise.
func (s *server) measureBacklog() {
	defer func() {
		if r := recover(); r != nil {
			log.Printf("panic while measuring the backlog: %v", r)
		}
	}()

	tokens, err := unmeasured(s.db)
	if err != nil {
		log.Printf("listing unmeasured submissions: %v", err)
		return
	}
	if len(tokens) == 0 {
		return
	}
	log.Printf("measuring %d submission(s) stored before they could be measured", len(tokens))

	for _, token := range tokens {
		audio, mime, err := audioByToken(s.db, token)
		if errors.Is(err, ErrNotFound) {
			// Deleted while this was running; nothing to measure.
			continue
		}
		if err != nil {
			log.Printf("measuring backlog: %v", err)
			continue
		}
		res, err := probeAudio(context.Background(), s.ffprobeBin, s.ffmpegBin, audio, extForMIME(mime))
		if err != nil {
			log.Printf("measuring backlog: %v", err)
			continue
		}
		if res.DurationMs <= 0 {
			continue
		}
		if res.Audio != nil {
			err = setRemuxedAudio(s.db, token, res.Audio, res.DurationMs)
		} else {
			err = setDuration(s.db, token, res.DurationMs)
		}
		if err != nil {
			log.Printf("measuring backlog: %v", err)
		}
	}
}

// extForMIME is the file extension ffmpeg needs to infer a container format.
// The MIME is the one sniffAudio decided from the bytes, so this is a mapping
// between two of our own labels rather than a guess about the file.
func extForMIME(mime string) string {
	switch mime {
	case "audio/mp4":
		return ".m4a"
	case "audio/ogg":
		return ".ogg"
	case "audio/flac":
		return ".flac"
	case "audio/mpeg":
		return ".mp3"
	case "audio/wav":
		return ".wav"
	case "audio/webm":
		return ".webm"
	}
	return ""
}

// notifyQuotaFull warns once per filling.
func (s *server) notifyQuotaFull(used int64) {
	if s.quotaWarned || !s.mail.enabled() {
		return
	}
	s.quotaWarned = true
	subject, text, html := quotaFullMail(used)
	go func() {
		if err := s.mail.notify(subject, text, html); err != nil {
			log.Printf("sending quota warning: %v", err)
		}
	}()
}

// refuse answers a rejected submission with a message meant to be read by the
// person who sent it.
func (s *server) refuse(w http.ResponseWriter, code int, msg string) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	body, _ := jsonString(msg)
	fmt.Fprintf(w, `{"ok":false,"error":%s}`+"\n", body)
}

// fail reports an internal error without telling the caller anything about it.
func (s *server) fail(w http.ResponseWriter, err error) {
	log.Printf("error: %v", err)
	http.Error(w, "Something went wrong here. Try again later.", http.StatusInternalServerError)
}

func readSecret(path string) (string, error) {
	b, err := os.ReadFile(path)
	if err != nil {
		return "", err
	}
	s := strings.TrimSpace(string(b))
	if s == "" {
		return "", errors.New("file is empty")
	}
	return s, nil
}

func isTruthy(s string) bool {
	switch strings.ToLower(strings.TrimSpace(s)) {
	case "1", "true", "yes", "on":
		return true
	}
	return false
}

func truncate(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n]
}

// sanitiseFilename keeps a display name without keeping anything that could be
// mistaken for a path. The name is only ever shown, never used to open a file,
// but it comes from a stranger and there is no reason to store the original.
func sanitiseFilename(name string) string {
	name = strings.TrimSpace(name)
	if i := strings.LastIndexAny(name, `/\`); i >= 0 {
		name = name[i+1:]
	}
	name = strings.Map(func(r rune) rune {
		if r < 0x20 || r == 0x7f {
			return -1
		}
		return r
	}, name)
	return truncate(name, 120)
}

// jsonString quotes a string for embedding in a JSON literal.
func jsonString(s string) (string, error) {
	var b strings.Builder
	b.WriteByte('"')
	for _, r := range s {
		switch r {
		case '"':
			b.WriteString(`\"`)
		case '\\':
			b.WriteString(`\\`)
		case '\n':
			b.WriteString(`\n`)
		case '\r':
			b.WriteString(`\r`)
		case '\t':
			b.WriteString(`\t`)
		default:
			if r < 0x20 {
				fmt.Fprintf(&b, `\u%04x`, r)
				continue
			}
			b.WriteRune(r)
		}
	}
	b.WriteByte('"')
	return b.String(), nil
}