1package replication23import (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"1617 "github.com/FelineStateMachine/tinyrelay/internal/event"18 "github.com/FelineStateMachine/tinyrelay/internal/storage"19 "github.com/FelineStateMachine/tinyrelay/internal/work"20)2122type Resolver interface {23 LookupNetIP(context.Context, string, string) ([]netip.Addr, error)24}2526type CallbackClient struct {27 HTTP *http.Client28 Resolver Resolver29 Timeout time.Duration30}3132func NewCallbackHandler(store *storage.Store, client CallbackClient) work.Handler {33 return NewCallbackHandlerWithPolicy(store, client, nil, nil, nil)34}3536func 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}3940// NewCallbackHandlerWithEventVisibility is the event-aware variant used by41// tenant adapters. It rechecks visibility against the exact event immediately42// before delivery, after registration revocation and filter matching.43func 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}4647func 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 err54 }55 if enabled != nil && !enabled() {56 return nil57 }58 registration, err := currentRegistration(ctx, store, envelope.Registration, policy)59 if err != nil || registration.ID == "" {60 return nil61 }62 envelope.Registration = registration63 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 err66 }67 if len(result.Events) == 0 {68 return nil69 }70 delivered := result.Events[0]71 if visible != nil && !visible(ctx, envelope.Registration.Owner) {72 return nil73 }74 if visibleEvent != nil && !visibleEvent(ctx, envelope.Registration.Owner, delivered) {75 return nil76 }77 if !PushMatches(envelope.Registration, delivered) {78 return nil79 }80 return client.Post(ctx, envelope.Registration, delivered)81 }82}8384func currentRegistration(ctx context.Context, store *storage.Store, expected PushRegistration, policy func() CallbackPolicy) (PushRegistration, error) {85 var raw string86 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{}, err88 }89 var item event.Event90 if err := json.Unmarshal([]byte(raw), &item); err != nil {91 return PushRegistration{}, err92 }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, nil104}105106func (c CallbackClient) Post(ctx context.Context, registration PushRegistration, e event.Event) error {107 body, err := CallbackPayload(registration, e)108 if err != nil {109 return err110 }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.HTTP116 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 err122 }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 nil134}135136// NewPinnedClient returns an HTTPS client that refuses private addresses and137// redirects, for calls to third-party services such as push endpoints.138func NewPinnedClient(timeout time.Duration) *http.Client { return pinnedClient(nil, timeout) }139140func pinnedClient(resolver Resolver, timeout time.Duration) *http.Client {141 if timeout <= 0 {142 timeout = 5 * time.Second143 }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, err150 }151 if resolver == nil {152 resolver = net.DefaultResolver153 }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 continue161 }162 conn, dialErr := (&net.Dialer{Timeout: timeout}).DialContext(ctx, network, net.JoinHostPort(ip.String(), port))163 if dialErr == nil {164 return conn, nil165 }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}172173func 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 true180 }181 }182 return false183}