| 1 | package replication |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/tls" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "net" |
| 11 | "net/http" |
| 12 | "net/netip" |
| 13 | "net/url" |
| 14 | "strings" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 18 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 19 | "github.com/FelineStateMachine/tinyrelay/internal/work" |
| 20 | ) |
| 21 | |
| 22 | type Resolver interface { |
| 23 | LookupNetIP(context.Context, string, string) ([]netip.Addr, error) |
| 24 | } |
| 25 | |
| 26 | type CallbackClient struct { |
| 27 | HTTP *http.Client |
| 28 | Resolver Resolver |
| 29 | Timeout time.Duration |
| 30 | } |
| 31 | |
| 32 | func NewCallbackHandler(store *storage.Store, client CallbackClient) work.Handler { |
| 33 | return NewCallbackHandlerWithPolicy(store, client, nil, nil, nil) |
| 34 | } |
| 35 | |
| 36 | func NewCallbackHandlerWithPolicy(store *storage.Store, client CallbackClient, policy func() CallbackPolicy, enabled func() bool, visible func(context.Context, string) bool) work.Handler { |
| 37 | return newCallbackHandler(store, client, policy, enabled, visible, nil) |
| 38 | } |
| 39 | |
| 40 | // NewCallbackHandlerWithEventVisibility is the event-aware variant used by |
| 41 | // tenant adapters. It rechecks visibility against the exact event immediately |
| 42 | // before delivery, after registration revocation and filter matching. |
| 43 | func NewCallbackHandlerWithEventVisibility(store *storage.Store, client CallbackClient, policy func() CallbackPolicy, enabled func() bool, visible func(context.Context, string) bool, visibleEvent func(context.Context, string, event.Event) bool) work.Handler { |
| 44 | return newCallbackHandler(store, client, policy, enabled, visible, visibleEvent) |
| 45 | } |
| 46 | |
| 47 | func newCallbackHandler(store *storage.Store, client CallbackClient, policy func() CallbackPolicy, enabled func() bool, visible func(context.Context, string) bool, visibleEvent func(context.Context, string, event.Event) bool) work.Handler { |
| 48 | return func(ctx context.Context, intent work.Intent) error { |
| 49 | var envelope struct { |
| 50 | Registration PushRegistration `json:"registration"` |
| 51 | } |
| 52 | if err := json.Unmarshal([]byte(intent.Payload), &envelope); err != nil { |
| 53 | return err |
| 54 | } |
| 55 | if enabled != nil && !enabled() { |
| 56 | return nil |
| 57 | } |
| 58 | registration, err := currentRegistration(ctx, store, envelope.Registration, policy) |
| 59 | if err != nil || registration.ID == "" { |
| 60 | return nil |
| 61 | } |
| 62 | envelope.Registration = registration |
| 63 | result, err := store.Query(ctx, event.Filter{IDs: []string{intent.EventID}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 1}) |
| 64 | if err != nil { |
| 65 | return err |
| 66 | } |
| 67 | if len(result.Events) == 0 { |
| 68 | return nil |
| 69 | } |
| 70 | delivered := result.Events[0] |
| 71 | if visible != nil && !visible(ctx, envelope.Registration.Owner) { |
| 72 | return nil |
| 73 | } |
| 74 | if visibleEvent != nil && !visibleEvent(ctx, envelope.Registration.Owner, delivered) { |
| 75 | return nil |
| 76 | } |
| 77 | if !PushMatches(envelope.Registration, delivered) { |
| 78 | return nil |
| 79 | } |
| 80 | return client.Post(ctx, envelope.Registration, delivered) |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | func currentRegistration(ctx context.Context, store *storage.Store, expected PushRegistration, policy func() CallbackPolicy) (PushRegistration, error) { |
| 85 | var raw string |
| 86 | if err := store.DB().QueryRowContext(ctx, `SELECT raw FROM events WHERE id=? AND kind=?` , expected.ID, event.KIND_PUSH_REGISTRATION).Scan(&raw); err != nil { |
| 87 | return PushRegistration{}, err |
| 88 | } |
| 89 | var item event.Event |
| 90 | if err := json.Unmarshal([]byte(raw), &item); err != nil { |
| 91 | return PushRegistration{}, err |
| 92 | } |
| 93 | approved := CallbackPolicy{} |
| 94 | if policy != nil { |
| 95 | approved = policy() |
| 96 | } else { |
| 97 | approved = CallbackPolicy{HostOrigins: []string{expected.Callback}, OwnerOrigins: []string{expected.Callback}} |
| 98 | } |
| 99 | parsed, err := ParsePushRegistration(item, approved, expected.Relay) |
| 100 | if err != nil || parsed.Owner != expected.Owner { |
| 101 | return PushRegistration{}, errors.New("replication: callback registration revoked" ) |
| 102 | } |
| 103 | return parsed, nil |
| 104 | } |
| 105 | |
| 106 | func (c CallbackClient) Post(ctx context.Context, registration PushRegistration, e event.Event) error { |
| 107 | body, err := CallbackPayload(registration, e) |
| 108 | if err != nil { |
| 109 | return err |
| 110 | } |
| 111 | parsed, err := url.Parse(registration.Callback) |
| 112 | if err != nil || parsed.Scheme != "https" || parsed.Hostname() == "" { |
| 113 | return errors.New("replication: callback URL must be HTTPS" ) |
| 114 | } |
| 115 | client := c.HTTP |
| 116 | if client == nil { |
| 117 | client = pinnedClient(c.Resolver, c.Timeout) |
| 118 | } |
| 119 | request, err := http.NewRequestWithContext(ctx, http.MethodPost, parsed.String(), strings.NewReader(string(body))) |
| 120 | if err != nil { |
| 121 | return err |
| 122 | } |
| 123 | request.Header.Set("content-type" , "application/json" ) |
| 124 | response, err := client.Do(request) |
| 125 | if err != nil { |
| 126 | return fmt.Errorf("replication: callback request: %w" , err) |
| 127 | } |
| 128 | defer response.Body.Close() |
| 129 | _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 4096)) |
| 130 | if response.StatusCode < 200 || response.StatusCode >= 300 { |
| 131 | return fmt.Errorf("replication: callback returned HTTP %d" , response.StatusCode) |
| 132 | } |
| 133 | return nil |
| 134 | } |
| 135 | |
| 136 | // NewPinnedClient returns an HTTPS client that refuses private addresses and |
| 137 | // redirects, for calls to third-party services such as push endpoints. |
| 138 | func NewPinnedClient(timeout time.Duration) *http.Client { return pinnedClient(nil, timeout) } |
| 139 | |
| 140 | func pinnedClient(resolver Resolver, timeout time.Duration) *http.Client { |
| 141 | if timeout <= 0 { |
| 142 | timeout = 5 * time.Second |
| 143 | } |
| 144 | transport := &http.Transport{ |
| 145 | TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12}, |
| 146 | DialContext: func(ctx context.Context, network, address string) (net.Conn, error) { |
| 147 | host, port, err := net.SplitHostPort(address) |
| 148 | if err != nil { |
| 149 | return nil, err |
| 150 | } |
| 151 | if resolver == nil { |
| 152 | resolver = net.DefaultResolver |
| 153 | } |
| 154 | ips, err := resolver.LookupNetIP(ctx, "ip" , host) |
| 155 | if err != nil || len(ips) == 0 { |
| 156 | return nil, errors.New("replication: callback host did not resolve" ) |
| 157 | } |
| 158 | for _, ip := range ips { |
| 159 | if isPrivateIP(ip) { |
| 160 | continue |
| 161 | } |
| 162 | conn, dialErr := (&net.Dialer{Timeout: timeout}).DialContext(ctx, network, net.JoinHostPort(ip.String(), port)) |
| 163 | if dialErr == nil { |
| 164 | return conn, nil |
| 165 | } |
| 166 | } |
| 167 | return nil, errors.New("replication: callback host resolved only to private addresses" ) |
| 168 | }, |
| 169 | } |
| 170 | return &http.Client{Transport: transport, Timeout: timeout, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} |
| 171 | } |
| 172 | |
| 173 | func isPrivateIP(addr netip.Addr) bool { |
| 174 | ip := net.IP(addr.AsSlice()) |
| 175 | private := []string{"10.0.0.0/8" , "100.64.0.0/10" , "127.0.0.0/8" , "169.254.0.0/16" , "172.16.0.0/12" , "192.0.0.0/24" , "192.168.0.0/16" , "::1/128" , "fc00::/7" , "fe80::/10" } |
| 176 | for _, raw := range private { |
| 177 | _, network, err := net.ParseCIDR(raw) |
| 178 | if err == nil && network.Contains(ip) { |
| 179 | return true |
| 180 | } |
| 181 | } |
| 182 | return false |
| 183 | } |