diff --git a/api/platform_client.go b/api/platform_client.go index 986e653..b8c57e7 100644 --- a/api/platform_client.go +++ b/api/platform_client.go @@ -15,16 +15,20 @@ import ( "log" "net/http" "net/url" + "reflect" "strconv" "strings" + "sync/atomic" "time" "browser.local/run/protocol" ) type PlatformClient struct { - baseURL string - httpClient *http.Client + baseURL string + httpClient *http.Client + serverClockOffsetNanos *atomic.Int64 + serverClockOffsetActive *atomic.Bool } type PlatformRequestError struct { @@ -134,7 +138,7 @@ func NewPlatformClientWithHTTPClient(rawURL string, httpClient *http.Client) (Pl httpClient = http.DefaultClient } - return PlatformClient{baseURL: strings.TrimRight(parsed.String(), "/"), httpClient: httpClient}, nil + return PlatformClient{baseURL: strings.TrimRight(parsed.String(), "/"), httpClient: httpClient, serverClockOffsetNanos: &atomic.Int64{}, serverClockOffsetActive: &atomic.Bool{}}, nil } func (c PlatformClient) BaseURL() string { @@ -166,7 +170,7 @@ func (c PlatformClient) StreamControlEvents(ctx context.Context, request protoco } httpRequest.Header.Set("Content-Type", "application/json") httpRequest.Header.Set("Accept", "text/event-stream") - signatureSummary, err := signRunRequest(httpRequest, body.Bytes()) + signatureSummary, err := signRunRequest(httpRequest, body.Bytes(), c.signatureTime()) if err != nil { return err } @@ -348,7 +352,7 @@ func postPlatformJSON[Request any, Response any](ctx context.Context, client Pla httpRequest.Header.Set("Content-Type", "application/json") httpRequest.Header.Set("Accept", "application/json") if path != "/api/v1/run/control/hello" { - signatureSummary, err := signRunRequest(httpRequest, body.Bytes()) + signatureSummary, err := signRunRequest(httpRequest, body.Bytes(), client.signatureTime()) if err != nil { log.Printf("RUN platform request status=sign_error method=POST base=%s path=%s durationMs=%d error=%s", diagnosticLogValue(client.baseURL), path, time.Since(startedAt).Milliseconds(), err) return response, err @@ -375,6 +379,7 @@ func postPlatformJSON[Request any, Response any](ctx context.Context, client Pla if err := json.NewDecoder(httpResponse.Body).Decode(&response); err != nil { return response, fmt.Errorf("decode platform response: %w", err) } + client.observeResponseServerTime(response) return response, nil } @@ -391,7 +396,7 @@ type runRequestSignatureSummary struct { Signature string } -func signRunRequest(request *http.Request, body []byte) (runRequestSignatureSummary, error) { +func signRunRequest(request *http.Request, body []byte, stamp time.Time) (runRequestSignatureSummary, error) { var envelope runRequestEnvelope if err := json.Unmarshal(body, &envelope); err != nil { return runRequestSignatureSummary{}, fmt.Errorf("decode Run signing envelope: %w", err) @@ -403,7 +408,7 @@ func signRunRequest(request *http.Request, body []byte) (runRequestSignatureSumm if _, err := rand.Read(nonceBytes); err != nil { return runRequestSignatureSummary{}, fmt.Errorf("create Run request nonce: %w", err) } - timestamp := strconv.FormatInt(time.Now().UTC().Unix(), 10) + timestamp := strconv.FormatInt(stamp.UTC().Unix(), 10) nonce := hex.EncodeToString(nonceBytes) bodyHash := sha256.Sum256(body) bodyHashHex := hex.EncodeToString(bodyHash[:]) @@ -418,6 +423,56 @@ func signRunRequest(request *http.Request, body []byte) (runRequestSignatureSumm return runRequestSignatureSummary{RunEndpointID: envelope.RunEndpointID, Timestamp: timestamp, Nonce: nonce, BodyHash: bodyHashHex, Signature: signature}, nil } +func (c PlatformClient) signatureTime() time.Time { + now := time.Now().UTC() + if c.serverClockOffsetActive != nil && c.serverClockOffsetActive.Load() && c.serverClockOffsetNanos != nil { + return now.Add(time.Duration(c.serverClockOffsetNanos.Load())) + } + return now +} + +func (c PlatformClient) observeResponseServerTime(response any) { + serverTime, ok := responseServerTime(response) + if !ok || serverTime.IsZero() || c.serverClockOffsetNanos == nil || c.serverClockOffsetActive == nil { + return + } + observedAt := time.Now().UTC() + offset := serverTime.UTC().Sub(observedAt) + previousActive := c.serverClockOffsetActive.Load() + previous := time.Duration(c.serverClockOffsetNanos.Load()) + c.serverClockOffsetNanos.Store(int64(offset)) + c.serverClockOffsetActive.Store(true) + if !previousActive || absDuration(offset-previous) > time.Second { + log.Printf("RUN platform clock status=calibrated offsetMs=%d serverTime=%s", offset.Milliseconds(), serverTime.UTC().Format(time.RFC3339)) + } +} + +func responseServerTime(response any) (time.Time, bool) { + value := reflect.ValueOf(response) + for value.Kind() == reflect.Pointer || value.Kind() == reflect.Interface { + if value.IsNil() { + return time.Time{}, false + } + value = value.Elem() + } + if value.Kind() != reflect.Struct { + return time.Time{}, false + } + field := value.FieldByName("ServerTime") + if !field.IsValid() || !field.CanInterface() { + return time.Time{}, false + } + serverTime, ok := field.Interface().(time.Time) + return serverTime, ok +} + +func absDuration(value time.Duration) time.Duration { + if value < 0 { + return -value + } + return value +} + func shortDiagnosticValue(value string) string { value = diagnosticLogValue(value) if value == "" { diff --git a/api/platform_client_test.go b/api/platform_client_test.go index 87cb334..ee0bd0d 100644 --- a/api/platform_client_test.go +++ b/api/platform_client_test.go @@ -136,6 +136,47 @@ func TestPlatformClientSignsRunChannelRequestsWithUniqueNonce(t *testing.T) { } } +func TestPlatformClientSignsWithObservedServerClock(t *testing.T) { + activeSessionToken := "session-token" + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/v1/run/control/hello": + writeTestJSON(t, w, protocol.RunHelloResponse{Accepted: true, RunEndpointID: "run-local", SessionToken: activeSessionToken, ServerTime: time.Now().UTC().Add(12 * time.Hour), HeartbeatIntervalSeconds: 15}) + case "/api/v1/run/control/heartbeat": + body, err := io.ReadAll(r.Body) + if err != nil { + t.Fatalf("read heartbeat body: %v", err) + } + verifyRunRequestSignature(t, r, body, "run-local", activeSessionToken) + timestamp, err := strconv.ParseInt(r.Header.Get("X-Run-Timestamp"), 10, 64) + if err != nil { + t.Fatalf("parse signed timestamp: %v", err) + } + signedAt := time.Unix(timestamp, 0).UTC() + serverNow := time.Now().UTC().Add(12 * time.Hour) + if delta := signedAt.Sub(serverNow); delta < -5*time.Second || delta > 5*time.Second { + t.Fatalf("expected signed timestamp to follow observed server clock, signedAt=%s serverNow=%s delta=%s", signedAt.Format(time.RFC3339), serverNow.Format(time.RFC3339), delta) + } + writeTestJSON(t, w, protocol.RunHeartbeatResponse{Accepted: true, RunEndpointID: "run-local", NextHeartbeatSeconds: 15, ServerTime: serverNow}) + default: + t.Fatalf("unexpected request path %s", r.URL.Path) + } + })) + defer server.Close() + + client, err := NewPlatformClient(server.URL) + if err != nil { + t.Fatalf("new client: %v", err) + } + hello, err := client.Hello(context.Background(), validRunHelloRequest()) + if err != nil { + t.Fatalf("hello: %v", err) + } + if _, err := client.Heartbeat(context.Background(), validRunHeartbeatRequest(hello.SessionToken)); err != nil { + t.Fatalf("heartbeat: %v", err) + } +} + func TestPlatformClientStreamsSignedControlEvents(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost || r.URL.Path != "/api/v1/run/control/events" {