418 lines
14 KiB
Go
418 lines
14 KiB
Go
package controlplane
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"crypto/tls"
|
|
"crypto/x509"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/klarkxy/nekonest-cloud/relay/internal/authsnapshot"
|
|
"github.com/klarkxy/nekonest/relaycore/protocol"
|
|
)
|
|
|
|
const maxControlResponse = 1 << 20
|
|
|
|
type Config struct {
|
|
BaseURL string
|
|
ClientCertFile string
|
|
ClientKeyFile string
|
|
CAFile string
|
|
Timeout time.Duration
|
|
HTTPClient *http.Client
|
|
}
|
|
|
|
type Client struct {
|
|
base *url.URL
|
|
http *http.Client
|
|
}
|
|
|
|
type RemoteError struct {
|
|
Status int
|
|
Body protocol.ServiceErrorPayload
|
|
}
|
|
|
|
func (e *RemoteError) Error() string {
|
|
if e.Body.Message != "" {
|
|
return fmt.Sprintf("control plane %s: %s", e.Body.ErrorCode, e.Body.Message)
|
|
}
|
|
return fmt.Sprintf("control plane HTTP %d", e.Status)
|
|
}
|
|
|
|
func exactBase(raw string) (*url.URL, error) {
|
|
parsed, err := url.Parse(strings.TrimSpace(raw))
|
|
if err != nil || parsed.Host == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" {
|
|
return nil, fmt.Errorf("invalid control-plane origin")
|
|
}
|
|
isLoopback := parsed.Hostname() == "localhost" || parsed.Hostname() == "127.0.0.1" || parsed.Hostname() == "::1"
|
|
if parsed.Scheme != "https" && !(parsed.Scheme == "http" && isLoopback) {
|
|
return nil, fmt.Errorf("control-plane origin must use HTTPS")
|
|
}
|
|
if parsed.Path != "" && parsed.Path != "/" {
|
|
return nil, fmt.Errorf("control-plane URL must be an origin")
|
|
}
|
|
parsed.Path = ""
|
|
return parsed, nil
|
|
}
|
|
|
|
func New(config Config) (*Client, error) {
|
|
base, err := exactBase(config.BaseURL)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
client := config.HTTPClient
|
|
if client == nil {
|
|
certificate, err := tls.LoadX509KeyPair(config.ClientCertFile, config.ClientKeyFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("load relay mTLS identity: %w", err)
|
|
}
|
|
caPEM, err := os.ReadFile(config.CAFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("load control-plane CA: %w", err)
|
|
}
|
|
roots := x509.NewCertPool()
|
|
if !roots.AppendCertsFromPEM(caPEM) {
|
|
return nil, fmt.Errorf("control-plane CA contains no certificates")
|
|
}
|
|
transport := http.DefaultTransport.(*http.Transport).Clone()
|
|
transport.TLSClientConfig = &tls.Config{
|
|
MinVersion: tls.VersionTLS13,
|
|
RootCAs: roots,
|
|
Certificates: []tls.Certificate{certificate},
|
|
}
|
|
timeout := config.Timeout
|
|
if timeout <= 0 {
|
|
timeout = 10 * time.Second
|
|
}
|
|
client = &http.Client{Transport: transport, Timeout: timeout}
|
|
}
|
|
clone := *client
|
|
clone.CheckRedirect = func(_ *http.Request, _ []*http.Request) error {
|
|
return http.ErrUseLastResponse
|
|
}
|
|
return &Client{base: base, http: &clone}, nil
|
|
}
|
|
|
|
func (c *Client) endpoint(path string) string {
|
|
return c.base.Scheme + "://" + c.base.Host + path
|
|
}
|
|
|
|
func (c *Client) doJSON(ctx context.Context, path string, request any, response any, headers http.Header) error {
|
|
body, err := json.Marshal(request)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.endpoint(path), bytes.NewReader(body))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req.Header.Set("Content-Type", "application/json")
|
|
req.Header.Set("Accept", "application/json")
|
|
for key, values := range headers {
|
|
for _, value := range values {
|
|
req.Header.Add(key, value)
|
|
}
|
|
}
|
|
res, err := c.http.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer res.Body.Close()
|
|
data, err := io.ReadAll(io.LimitReader(res.Body, maxControlResponse+1))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(data) > maxControlResponse {
|
|
return fmt.Errorf("control-plane response exceeds limit")
|
|
}
|
|
if res.StatusCode < 200 || res.StatusCode >= 300 {
|
|
var envelope protocol.ServiceErrorPayload
|
|
if err := json.Unmarshal(data, &envelope); err != nil || envelope.ErrorCode == "" {
|
|
return &RemoteError{Status: res.StatusCode, Body: protocol.ServiceErrorPayload{
|
|
ErrorCode: "route_unavailable", Message: "Control plane rejected the request", Retryable: false,
|
|
}}
|
|
}
|
|
return &RemoteError{Status: res.StatusCode, Body: envelope}
|
|
}
|
|
if response == nil {
|
|
return nil
|
|
}
|
|
decoder := json.NewDecoder(bytes.NewReader(data))
|
|
decoder.DisallowUnknownFields()
|
|
if err := decoder.Decode(response); err != nil {
|
|
return fmt.Errorf("decode control-plane response: %w", err)
|
|
}
|
|
if decoder.Decode(&struct{}{}) != io.EOF {
|
|
return fmt.Errorf("control-plane response contains trailing JSON")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func SHA256Hex(value string) string {
|
|
digest := sha256.Sum256([]byte(value))
|
|
return hex.EncodeToString(digest[:])
|
|
}
|
|
|
|
type SnapshotResponse struct {
|
|
Snapshot authsnapshot.Signed `json:"snapshot"`
|
|
PublicKeyJWK json.RawMessage `json:"public_key_jwk"`
|
|
}
|
|
|
|
type RouteResolution struct {
|
|
RelayNodeID string `json:"relay_node_id"`
|
|
PlacementGeneration int64 `json:"placement_generation"`
|
|
HomeRegion string `json:"home_region"`
|
|
Local bool `json:"local"`
|
|
EndpointRef string `json:"endpoint_ref,omitempty"`
|
|
}
|
|
|
|
func (route RouteResolution) Validate() error {
|
|
if route.RelayNodeID == "" || route.PlacementGeneration < 1 || route.HomeRegion == "" {
|
|
return errors.New("control plane returned an incomplete route")
|
|
}
|
|
if route.Local && route.EndpointRef != "" {
|
|
return errors.New("local route unexpectedly contains an endpoint reference")
|
|
}
|
|
if !route.Local && route.EndpointRef == "" {
|
|
return errors.New("remote route contains no endpoint reference")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *Client) ResolveDeviceRoute(ctx context.Context, deviceID, tokenHash string) (RouteResolution, error) {
|
|
var response RouteResolution
|
|
err := c.doJSON(ctx, "/api/internal/relay/resolve-device-route", map[string]string{
|
|
"device_id": deviceID, "token_hash": tokenHash,
|
|
}, &response, nil)
|
|
if err == nil {
|
|
err = response.Validate()
|
|
}
|
|
return response, err
|
|
}
|
|
|
|
func (c *Client) ResolvePhoneRoute(ctx context.Context, routeHandle, phoneTokenHash string) (RouteResolution, error) {
|
|
var response RouteResolution
|
|
err := c.doJSON(ctx, "/api/internal/relay/resolve-phone-route", map[string]string{
|
|
"route_handle": routeHandle, "phone_token_hash": phoneTokenHash,
|
|
}, &response, nil)
|
|
if err == nil {
|
|
err = response.Validate()
|
|
}
|
|
return response, err
|
|
}
|
|
|
|
func (c *Client) ResolveTenantRoute(ctx context.Context, tenantID string, generation int64) (RouteResolution, error) {
|
|
var response RouteResolution
|
|
err := c.doJSON(ctx, "/api/internal/relay/resolve-tenant-route", map[string]any{
|
|
"tenant_id": tenantID, "placement_generation": generation,
|
|
}, &response, nil)
|
|
if err == nil {
|
|
err = response.Validate()
|
|
}
|
|
return response, err
|
|
}
|
|
|
|
func (c *Client) ResolveHandoffRoute(ctx context.Context, ticket, pwaOrigin string) (RouteResolution, error) {
|
|
var response RouteResolution
|
|
err := c.doJSON(ctx, "/api/internal/relay/resolve-handoff-route", map[string]string{
|
|
"ticket": ticket, "pwa_origin": pwaOrigin,
|
|
}, &response, nil)
|
|
if err == nil {
|
|
err = response.Validate()
|
|
}
|
|
return response, err
|
|
}
|
|
|
|
func (c *Client) AuthorizeDevice(ctx context.Context, deviceID, tokenHash string) (SnapshotResponse, error) {
|
|
var response SnapshotResponse
|
|
err := c.doJSON(ctx, "/api/internal/relay/authorize-device", map[string]string{
|
|
"device_id": deviceID, "token_hash": tokenHash,
|
|
}, &response, nil)
|
|
return response, err
|
|
}
|
|
|
|
func (c *Client) AuthorizationSnapshot(ctx context.Context, tenantID string, generation int64) (SnapshotResponse, error) {
|
|
var response SnapshotResponse
|
|
err := c.doJSON(ctx, "/api/internal/relay/authorization-snapshot", map[string]any{
|
|
"tenant_id": tenantID, "placement_generation": generation,
|
|
}, &response, nil)
|
|
return response, err
|
|
}
|
|
|
|
type DeltaResponse struct {
|
|
TenantID string `json:"tenant_id"`
|
|
AuthorizationRevision int64 `json:"authorization_revision"`
|
|
TenantStatus string `json:"tenant_status"`
|
|
Changed bool `json:"changed"`
|
|
CheckedAt string `json:"checked_at"`
|
|
}
|
|
|
|
func (c *Client) AuthorizationDelta(ctx context.Context, tenantID string, afterRevision int64) (DeltaResponse, error) {
|
|
var response DeltaResponse
|
|
err := c.doJSON(ctx, "/api/internal/relay/authorization-delta", map[string]any{
|
|
"tenant_id": tenantID, "after_revision": afterRevision,
|
|
}, &response, nil)
|
|
return response, err
|
|
}
|
|
|
|
type MigrationAssignment struct {
|
|
MigrationID string `json:"migration_id"`
|
|
TenantID string `json:"tenant_id"`
|
|
Role string `json:"role"`
|
|
SourceNodeID string `json:"source_node_id"`
|
|
TargetNodeID string `json:"target_node_id"`
|
|
SourceGeneration int64 `json:"source_generation"`
|
|
TargetGeneration int64 `json:"target_generation"`
|
|
State string `json:"state"`
|
|
BackupRef string `json:"backup_ref,omitempty"`
|
|
ManifestSHA256 string `json:"manifest_sha256,omitempty"`
|
|
FinalizeAfter string `json:"finalize_after,omitempty"`
|
|
}
|
|
|
|
type PurgeAssignment struct {
|
|
PurgeID string `json:"purge_id"`
|
|
TenantID string `json:"tenant_id"`
|
|
PlacementGeneration int64 `json:"placement_generation"`
|
|
}
|
|
|
|
type HeartbeatAssignments struct {
|
|
Migrations []MigrationAssignment
|
|
Purges []PurgeAssignment
|
|
}
|
|
|
|
func (c *Client) Heartbeat(ctx context.Context, generation int64, capacityTenants int) (HeartbeatAssignments, error) {
|
|
var response struct {
|
|
Accepted bool `json:"accepted"`
|
|
CheckedAt string `json:"checked_at"`
|
|
Migrations []MigrationAssignment `json:"migrations"`
|
|
Purges []PurgeAssignment `json:"purges"`
|
|
}
|
|
err := c.doJSON(ctx, "/api/internal/relay/heartbeat", map[string]any{
|
|
"generation": generation, "capacity_tenants": capacityTenants,
|
|
}, &response, nil)
|
|
if err == nil && !response.Accepted {
|
|
return HeartbeatAssignments{}, errors.New("control plane did not accept relay heartbeat")
|
|
}
|
|
return HeartbeatAssignments{Migrations: response.Migrations, Purges: response.Purges}, err
|
|
}
|
|
|
|
type MigrationAdvance struct {
|
|
MigrationID string `json:"migration_id"`
|
|
Action string `json:"action"`
|
|
BackupRef string `json:"backup_ref,omitempty"`
|
|
ManifestSHA256 string `json:"manifest_sha256,omitempty"`
|
|
ErrorCode string `json:"error_code,omitempty"`
|
|
}
|
|
|
|
func (c *Client) AdvanceMigration(ctx context.Context, input MigrationAdvance) error {
|
|
var response struct {
|
|
MigrationID string `json:"migration_id"`
|
|
State string `json:"state"`
|
|
}
|
|
return c.doJSON(ctx, "/api/internal/relay/migrations/advance", input, &response, nil)
|
|
}
|
|
|
|
type PurgeAdvance struct {
|
|
PurgeID string `json:"purge_id"`
|
|
Action string `json:"action"`
|
|
EvidenceSHA256 string `json:"evidence_sha256,omitempty"`
|
|
ErrorCode string `json:"error_code,omitempty"`
|
|
}
|
|
|
|
func (c *Client) AdvancePurge(ctx context.Context, input PurgeAdvance) error {
|
|
var response struct {
|
|
PurgeID string `json:"purge_id"`
|
|
State string `json:"state"`
|
|
}
|
|
return c.doJSON(ctx, "/api/internal/relay/purges/advance", input, &response, nil)
|
|
}
|
|
|
|
type PhoneAuthorization struct {
|
|
TenantID string `json:"tenant_id"`
|
|
HomeRegion string `json:"home_region"`
|
|
RelayNodeID string `json:"relay_node_id"`
|
|
PlacementGeneration int64 `json:"placement_generation"`
|
|
Phone struct {
|
|
PhoneID string `json:"phone_id"`
|
|
Name string `json:"name"`
|
|
Ed25519Public string `json:"ed25519_public"`
|
|
X25519Public string `json:"x25519_public"`
|
|
IdentityFingerprint string `json:"identity_fingerprint"`
|
|
} `json:"phone"`
|
|
}
|
|
|
|
func (c *Client) AuthorizePhone(ctx context.Context, routeHandle, phoneTokenHash string) (PhoneAuthorization, error) {
|
|
var response PhoneAuthorization
|
|
err := c.doJSON(ctx, "/api/internal/relay/authorize-phone", map[string]string{
|
|
"route_handle": routeHandle, "phone_token_hash": phoneTokenHash,
|
|
}, &response, nil)
|
|
return response, err
|
|
}
|
|
|
|
type ConsumedHandoff struct {
|
|
HandoffID string `json:"handoff_id"`
|
|
TenantID string `json:"tenant_id"`
|
|
Name string `json:"name"`
|
|
PhoneEd25519Public string `json:"phone_ed25519_public"`
|
|
PhoneX25519Public string `json:"phone_x25519_public"`
|
|
IdentityFingerprint string `json:"identity_fingerprint"`
|
|
PlacementGeneration int64 `json:"placement_generation"`
|
|
}
|
|
|
|
func (c *Client) ConsumePhoneHandoff(ctx context.Context, request any) (ConsumedHandoff, error) {
|
|
var response ConsumedHandoff
|
|
err := c.doJSON(ctx, "/api/internal/relay/consume-phone-handoff", request, &response, nil)
|
|
return response, err
|
|
}
|
|
|
|
func (c *Client) CompletePhoneHandoff(ctx context.Context, handoffID, phoneID, phoneTokenHash, routeHandleHash string) error {
|
|
var response struct {
|
|
Completed bool `json:"completed"`
|
|
}
|
|
err := c.doJSON(ctx, "/api/internal/relay/complete-phone-handoff", map[string]string{
|
|
"handoff_id": handoffID,
|
|
"phone_id": phoneID,
|
|
"phone_token_hash": phoneTokenHash,
|
|
"route_handle_hash": routeHandleHash,
|
|
}, &response, nil)
|
|
if err == nil && !response.Completed {
|
|
return errors.New("control plane did not complete phone handoff")
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (c *Client) RevokePhone(ctx context.Context, tenantID, phoneID, reason string) error {
|
|
var response struct {
|
|
Revoked bool `json:"revoked"`
|
|
}
|
|
err := c.doJSON(ctx, "/api/internal/relay/revoke-phone", map[string]string{
|
|
"tenant_id": tenantID, "phone_id": phoneID, "reason": reason,
|
|
}, &response, nil)
|
|
if err == nil && !response.Revoked {
|
|
return errors.New("control plane did not revoke phone")
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (c *Client) RegisterDevice(ctx context.Context, bootstrap, sourceHash string, request json.RawMessage) (protocol.DeviceRegistrationResponse, error) {
|
|
var decoded map[string]any
|
|
if err := json.Unmarshal(request, &decoded); err != nil {
|
|
return protocol.DeviceRegistrationResponse{}, errors.New("invalid registration JSON")
|
|
}
|
|
decoded["bootstrap_token"] = bootstrap
|
|
decoded["source_hash"] = sourceHash
|
|
var response protocol.DeviceRegistrationResponse
|
|
err := c.doJSON(ctx, "/api/internal/relay/register-device", decoded, &response, nil)
|
|
return response, err
|
|
}
|