this repo has no description
split out forwarding helper to separate file; docstring
+92
-76
+91
cmd/relay/forward.go
+91
cmd/relay/forward.go
···
1
+
package main
2
+
3
+
import (
4
+
"bytes"
5
+
"io"
6
+
"net/http"
7
+
"strings"
8
+
"time"
9
+
10
+
"github.com/bluesky-social/indigo/cmd/relay/relay"
11
+
12
+
"github.com/labstack/echo/v4"
13
+
)
14
+
15
+
// Forwards HTTP request on to sibling relay instances, if they are configured.
16
+
//
17
+
// This method expects to be run "in the background" as a goroutine, so it doesn't take a `context.Context`, and does not return an `error`. It logs both success and failure. The `echo.Context` request has presumably been read, so any body is passed separately. The `echo.Context` response is presumably being returned or already finalized concurrently.
18
+
func (s *Service) ForwardSiblingRequest(c echo.Context, body []byte) {
19
+
20
+
if len(s.config.SiblingRelayHosts) == 0 {
21
+
return
22
+
}
23
+
24
+
// if the request itself was already forwarded, or user-agent seems to be a relay, then don't forward on further (to prevent loops)
25
+
req := c.Request()
26
+
for _, via := range req.Header.Values("Via") {
27
+
if strings.Contains(via, "atproto-relay") {
28
+
s.logger.Info("not re-forwarding request to sibling relay", "header", "Via", "value", via)
29
+
return
30
+
}
31
+
}
32
+
for _, ua := range req.Header.Values("User-Agent") {
33
+
if strings.Contains(ua, "atproto-relay") {
34
+
s.logger.Info("not re-forwarding request to sibling relay", "header", "User-Agent", "value", ua)
35
+
return
36
+
}
37
+
}
38
+
39
+
// TODO: could turn this in to a shared/persistent client
40
+
client := http.Client{
41
+
Timeout: 10 * time.Second,
42
+
}
43
+
44
+
for _, rawHost := range s.config.SiblingRelayHosts {
45
+
hostname, noSSL, err := relay.ParseHostname(rawHost)
46
+
if err != nil {
47
+
s.logger.Error("invalid sibling hostname configured", "host", rawHost, "err", err)
48
+
return
49
+
}
50
+
u := req.URL
51
+
u.Host = hostname
52
+
if noSSL {
53
+
u.Scheme = "http"
54
+
} else {
55
+
u.Scheme = "https"
56
+
}
57
+
var b io.Reader
58
+
if body != nil {
59
+
b = bytes.NewBuffer(body)
60
+
}
61
+
upstreamReq, err := http.NewRequest(req.Method, u.String(), b)
62
+
if err != nil {
63
+
s.logger.Error("creating admin forward request failed", "method", req.Method, "url", u.String(), "err", err)
64
+
continue
65
+
}
66
+
67
+
// copy some headers from inbound request
68
+
for _, hdr := range []string{"Accept", "User-Agent", "Authorization", "Content-Type"} {
69
+
val := req.Header.Get(hdr)
70
+
if val != "" {
71
+
upstreamReq.Header.Set(hdr, val)
72
+
}
73
+
}
74
+
75
+
// add Via header (critical to prevent forwarding loops)
76
+
upstreamReq.Header.Add("Via", s.relay.Config.UserAgent)
77
+
78
+
upstreamResp, err := client.Do(upstreamReq)
79
+
if err != nil {
80
+
s.logger.Warn("forwarded admin HTTP request failed", "method", req.Method, "sibling", hostname, "url", u.String(), "err", err)
81
+
continue
82
+
}
83
+
if upstreamResp.StatusCode != http.StatusOK {
84
+
respBytes, _ := io.ReadAll(upstreamResp.Body)
85
+
s.logger.Warn("forwarded admin HTTP request failed", "method", req.Method, "sibling", hostname, "url", u.String(), "statusCode", upstreamResp.StatusCode, "body", string(respBytes))
86
+
continue
87
+
}
88
+
upstreamResp.Body.Close()
89
+
s.logger.Info("successfully forwarded admin HTTP request", "method", req.Method, "url", u.String())
90
+
}
91
+
}
+1
-76
cmd/relay/handlers_admin.go
+1
-76
cmd/relay/handlers_admin.go
···
1
1
package main
2
2
3
3
import (
4
-
"bytes"
5
4
"encoding/json"
6
5
"errors"
7
6
"fmt"
8
-
"io"
9
7
"net/http"
10
8
"strconv"
11
9
"strings"
···
20
18
dto "github.com/prometheus/client_model/go"
21
19
)
22
20
23
-
// this is the same as the regular com.atproto.sync.requestCrawl endpoint, except it sets a flag to bypass configuration checks
21
+
// This endpint is basically the same as the regular com.atproto.sync.requestCrawl endpoint, except it sets a flag to bypass configuration checks.
24
22
func (s *Service) handleAdminRequestCrawl(c echo.Context) error {
25
23
var body comatproto.SyncRequestCrawl_Input
26
24
if err := c.Bind(&body); err != nil {
···
514
512
"success": "true",
515
513
})
516
514
}
517
-
518
-
// this method expects to be run in a goroutine. it does not take a `context.Context`, the input `echo.Context` has likely be cancelled/closed, and does not return an error (only logs)
519
-
func (s *Service) ForwardSiblingRequest(c echo.Context, body []byte) {
520
-
521
-
if len(s.config.SiblingRelayHosts) == 0 {
522
-
return
523
-
}
524
-
525
-
// if this request was forwarded, or user-agent matches, then don't forward
526
-
req := c.Request()
527
-
for _, via := range req.Header.Values("Via") {
528
-
if strings.Contains(via, "atproto-relay") {
529
-
s.logger.Info("not re-forwarding request to sibling relay", "header", "Via", "value", via)
530
-
return
531
-
}
532
-
}
533
-
for _, ua := range req.Header.Values("User-Agent") {
534
-
if strings.Contains(ua, "atproto-relay") {
535
-
s.logger.Info("not re-forwarding request to sibling relay", "header", "User-Agent", "value", ua)
536
-
return
537
-
}
538
-
}
539
-
540
-
client := http.Client{
541
-
Timeout: 10 * time.Second,
542
-
}
543
-
544
-
for _, rawHost := range s.config.SiblingRelayHosts {
545
-
hostname, noSSL, err := relay.ParseHostname(rawHost)
546
-
if err != nil {
547
-
s.logger.Error("invalid sibling hostname configured", "host", rawHost, "err", err)
548
-
return
549
-
}
550
-
u := req.URL
551
-
u.Host = hostname
552
-
if noSSL {
553
-
u.Scheme = "http"
554
-
} else {
555
-
u.Scheme = "https"
556
-
}
557
-
var b io.Reader
558
-
if body != nil {
559
-
b = bytes.NewBuffer(body)
560
-
}
561
-
upstreamReq, err := http.NewRequest(req.Method, u.String(), b)
562
-
if err != nil {
563
-
s.logger.Error("creating admin forward request failed", "method", req.Method, "url", u.String(), "err", err)
564
-
continue
565
-
}
566
-
567
-
// copy some headers from inbound request
568
-
for _, hdr := range []string{"Accept", "User-Agent", "Authorization", "Content-Type"} {
569
-
val := req.Header.Get(hdr)
570
-
if val != "" {
571
-
upstreamReq.Header.Set(hdr, val)
572
-
}
573
-
}
574
-
upstreamReq.Header.Add("Via", s.relay.Config.UserAgent)
575
-
576
-
upstreamResp, err := client.Do(upstreamReq)
577
-
if err != nil {
578
-
s.logger.Error("forwarded admin HTTP request failed", "method", req.Method, "url", u.String(), "err", err)
579
-
continue
580
-
}
581
-
if upstreamResp.StatusCode != http.StatusOK {
582
-
respBytes, _ := io.ReadAll(upstreamResp.Body)
583
-
s.logger.Error("forwarded admin HTTP request failed", "method", req.Method, "url", u.String(), "statusCode", upstreamResp.StatusCode, "body", string(respBytes))
584
-
continue
585
-
}
586
-
upstreamResp.Body.Close()
587
-
s.logger.Info("successfully forwarded admin HTTP request", "method", req.Method, "url", u.String())
588
-
}
589
-
}