From 86e36b80cae6ac6251f096ce68bfbb61bfe52119 Mon Sep 17 00:00:00 2001 From: kinzhi Date: Tue, 21 Jul 2026 14:57:09 +0800 Subject: [PATCH] chore: sync automatic Runtime credentials --- README.md | 39 +- README.zh-CN.md | 36 +- internal/agentnode/node.go | 5 + .../openlinker-go/runtime_a2a_proxy.go | 35 ++ .../openlinker-go/runtime_worker.go | 61 ++- .../openlinker-go/runtime_worker_client.go | 117 ++++- .../runtime_worker_credentials.go | 487 ++++++++++++++++++ .../openlinker-go/runtime_worker_discovery.go | 42 +- .../openlinker-go/runtime_worker_transport.go | 29 +- 9 files changed, 746 insertions(+), 105 deletions(-) create mode 100644 vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_credentials.go diff --git a/README.md b/README.md index 235b22e..7ea8b4a 100644 --- a/README.md +++ b/README.md @@ -79,9 +79,7 @@ source with the commands below. Prerequisites: - Go 1.25 or newer -- an Agent and Node registered in Core -- their lowercase UUIDs and an Agent Token -- a Core-issued client certificate, private key, and trusted CA bundle +- an active Agent Token - a private, persistent data directory - a local backend @@ -92,35 +90,16 @@ go test ./... go build ./cmd/openlinker-agent-node ``` -Enroll the Runtime Node with this Adapter's exact implementation version. Core -rejects a Session when the enrolled version and Worker hello differ. Set -`NODE_VERSION` to the complete value reported by the binary release, or copy -the complete `AgentNodeVersion` string from `internal/agentnode/node.go` for a -source build: - -```bash -NODE_VERSION=openlinker-agent-node/0.x.y -DATABASE_URL='postgres://...' ./api runtime-node issue \ - --ca-cert /secure/runtime-client-ca.crt \ - --ca-key /secure/runtime-client-ca.key \ - --display-name 'legacy-backend-adapter' \ - --node-version "${NODE_VERSION}" \ - --capacity 1 \ - --cert-out /run/openlinker/node.crt \ - --key-out /run/openlinker/node.key -``` +On first start the SDK creates the Node ID and P-256 private key inside the +private data directory, binds that public key one-to-one to the Agent Token, +and obtains a 24-hour client certificate. Renewal is automatic. Run a local HTTP backend: ```bash OPENLINKER_URL=https://openlinker.example \ -OPENLINKER_NODE_ID=11111111-1111-4111-8111-111111111111 \ -OPENLINKER_AGENT_ID=22222222-2222-4222-8222-222222222222 \ OPENLINKER_AGENT_TOKEN=ol_agent_xxx \ OPENLINKER_AGENT_NODE_DATA_DIR=/var/lib/openlinker-agent-node \ -OPENLINKER_AGENT_NODE_MTLS_CERT_FILE=/run/openlinker/node.crt \ -OPENLINKER_AGENT_NODE_MTLS_KEY_FILE=/run/openlinker/node.key \ -OPENLINKER_AGENT_NODE_MTLS_CA_FILE=/run/openlinker/core-ca.crt \ OPENLINKER_AGENT_NODE_TRANSPORT=auto \ OPENLINKER_AGENT_NODE_ADAPTER=http \ OPENLINKER_AGENT_NODE_HTTP_URL=http://127.0.0.1:18080/run \ @@ -151,13 +130,13 @@ stops startup instead of falling back to the ordinary API origin. | Variable | Purpose | | --- | --- | | `OPENLINKER_URL` | OpenLinker platform origin used to discover the Runtime connection | -| `OPENLINKER_NODE_ID` | Registered Node UUID | -| `OPENLINKER_AGENT_ID` | Agent UUID served by this process | +| `OPENLINKER_NODE_ID` | Optional legacy Node UUID override; generated automatically by default | +| `OPENLINKER_AGENT_ID` | Optional legacy Agent UUID override; resolved from the Agent Token by default | | `OPENLINKER_AGENT_TOKEN` | Long-lived Agent Token kept inside the node | | `OPENLINKER_AGENT_NODE_DATA_DIR` | Directory selected for the SDK `FileRuntimeStore` | -| `OPENLINKER_AGENT_NODE_MTLS_CERT_FILE` | Client certificate | -| `OPENLINKER_AGENT_NODE_MTLS_KEY_FILE` | Client private key | -| `OPENLINKER_AGENT_NODE_MTLS_CA_FILE` | CA bundle used to verify Core | +| `OPENLINKER_AGENT_NODE_MTLS_CERT_FILE` | Optional external-PKI compatibility certificate | +| `OPENLINKER_AGENT_NODE_MTLS_KEY_FILE` | Optional external-PKI compatibility private key | +| `OPENLINKER_AGENT_NODE_MTLS_CA_FILE` | Optional external-PKI compatibility CA bundle | | `OPENLINKER_AGENT_NODE_MTLS_SERVER_NAME` | Optional certificate server-name override | | `OPENLINKER_AGENT_NODE_TRANSPORT` | `auto` (default), `ws`, or `pull`; all share one Runtime session | diff --git a/README.zh-CN.md b/README.zh-CN.md index 4996ea8..29e01a1 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -67,9 +67,7 @@ Linux、macOS、Windows 预构建二进制及相邻的 `.sha256` 文件发布在 需要准备： - Go 1.25 或更高版本 -- 已在 Core 注册的 Agent 和 Node -- 两者的小写 UUID 与 Agent Token -- Core 签发的 client certificate、private key 和受信 CA bundle +- 有效的 Agent Token - 私有、可持久化的数据目录 - 本地 backend @@ -80,33 +78,15 @@ go test ./... go build ./cmd/openlinker-agent-node ``` -登记 Runtime Node 时必须写入当前 Adapter 的精确实现版本。登记值与 Worker hello 不一致时， -Core 会拒绝 Session。`NODE_VERSION` 应填写二进制 release 报告的完整值；源码构建则原样 -复制 `internal/agentnode/node.go` 中完整的 `AgentNodeVersion` 字符串： - -```bash -NODE_VERSION=openlinker-agent-node/0.x.y -DATABASE_URL='postgres://...' ./api runtime-node issue \ - --ca-cert /secure/runtime-client-ca.crt \ - --ca-key /secure/runtime-client-ca.key \ - --display-name 'legacy-backend-adapter' \ - --node-version "${NODE_VERSION}" \ - --capacity 1 \ - --cert-out /run/openlinker/node.crt \ - --key-out /run/openlinker/node.key -``` +首次启动时，SDK 会在私有目录生成 Node ID 和 P-256 私钥，将公钥一对一绑定到 Agent Token， +并取得 24 小时客户端证书；续期全自动完成。 运行本地 HTTP backend： ```bash OPENLINKER_URL=https://openlinker.example \ -OPENLINKER_NODE_ID=11111111-1111-4111-8111-111111111111 \ -OPENLINKER_AGENT_ID=22222222-2222-4222-8222-222222222222 \ OPENLINKER_AGENT_TOKEN=ol_agent_xxx \ OPENLINKER_AGENT_NODE_DATA_DIR=/var/lib/openlinker-agent-node \ -OPENLINKER_AGENT_NODE_MTLS_CERT_FILE=/run/openlinker/node.crt \ -OPENLINKER_AGENT_NODE_MTLS_KEY_FILE=/run/openlinker/node.key \ -OPENLINKER_AGENT_NODE_MTLS_CA_FILE=/run/openlinker/core-ca.crt \ OPENLINKER_AGENT_NODE_TRANSPORT=auto \ OPENLINKER_AGENT_NODE_ADAPTER=http \ OPENLINKER_AGENT_NODE_HTTP_URL=http://127.0.0.1:18080/run \ @@ -133,13 +113,13 @@ SDK 加密 spool 的上限是 512 MiB 和 10,000 条记录。使用量达到 80% | 环境变量 | 用途 | | --- | --- | | `OPENLINKER_URL` | OpenLinker 平台地址，用于自动发现 Runtime 连接信息 | -| `OPENLINKER_NODE_ID` | 已注册 Node 的 UUID | -| `OPENLINKER_AGENT_ID` | 当前进程承载的 Agent UUID | +| `OPENLINKER_NODE_ID` | 可选的旧版 Node UUID 覆盖；默认自动生成 | +| `OPENLINKER_AGENT_ID` | 可选的旧版 Agent UUID 覆盖；默认从 Agent Token 解析 | | `OPENLINKER_AGENT_TOKEN` | 只保留在节点内的长效 Agent Token | | `OPENLINKER_AGENT_NODE_DATA_DIR` | 交给 SDK `FileRuntimeStore` 的目录 | -| `OPENLINKER_AGENT_NODE_MTLS_CERT_FILE` | client certificate | -| `OPENLINKER_AGENT_NODE_MTLS_KEY_FILE` | client private key | -| `OPENLINKER_AGENT_NODE_MTLS_CA_FILE` | 用来校验 Core 的 CA bundle | +| `OPENLINKER_AGENT_NODE_MTLS_CERT_FILE` | 可选的外部 PKI 兼容证书 | +| `OPENLINKER_AGENT_NODE_MTLS_KEY_FILE` | 可选的外部 PKI 兼容私钥 | +| `OPENLINKER_AGENT_NODE_MTLS_CA_FILE` | 可选的外部 PKI 兼容 CA bundle | | `OPENLINKER_AGENT_NODE_MTLS_SERVER_NAME` | 可选的证书 server name 覆盖值 | | `OPENLINKER_AGENT_NODE_TRANSPORT` | `auto`（默认）、`ws` 或 `pull`；三者共用同一 Runtime session | diff --git a/internal/agentnode/node.go b/internal/agentnode/node.go index ad1305d..908a935 100644 --- a/internal/agentnode/node.go +++ b/internal/agentnode/node.go @@ -151,6 +151,11 @@ func (node *Node) newPublicA2AProxy(ctx context.Context) (publicA2AProxy, error) RuntimeURL: node.RuntimeURL, AgentToken: node.AgentToken, AgentSlug: node.PublicA2A.Slug, + DataDir: node.DataDir, + NodeID: node.NodeID, + AgentID: node.AgentID, + NodeVersion: AgentNodeVersion, + Capacity: node.Capacity, MTLS: openlinker.RuntimeMTLSConfig{ CertFile: node.MTLSCertFile, KeyFile: node.MTLSKeyFile, diff --git a/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_a2a_proxy.go b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_a2a_proxy.go index 6b3c68d..c6c1b20 100644 --- a/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_a2a_proxy.go +++ b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_a2a_proxy.go @@ -30,6 +30,11 @@ type RuntimeA2AProxyConfig struct { RuntimeURL string AgentToken string AgentSlug string + DataDir string + NodeID string + AgentID string + NodeVersion string + Capacity int64 MTLS RuntimeMTLSConfig } @@ -55,6 +60,36 @@ func NewRuntimeA2AProxy(ctx context.Context, config RuntimeA2AProxyConfig) (*Run if err != nil { return nil, err } + explicitMTLS := config.MTLS.CertFile != "" && config.MTLS.KeyFile != "" && config.MTLS.CAFile != "" + if !explicitMTLS || !connection.MTLSRequired { + credentialEndpoint := connection.CredentialEndpoint + if credentialEndpoint == "" && config.PlatformURL != "" { + platformOrigin, platformErr := validatePlatformOrigin(config.PlatformURL) + if platformErr != nil { + return nil, platformErr + } + credentialEndpoint = platformOrigin + "/api/v1/runtime-credentials" + } + manager, managerErr := newRuntimeCredentialManager( + config.DataDir, credentialEndpoint, config.AgentToken, + config.NodeID, config.AgentID, config.NodeVersion, config.Capacity, nil, + ) + if managerErr != nil { + return nil, managerErr + } + if managerErr = manager.Ensure(ctx, false); managerErr != nil { + return nil, managerErr + } + config.MTLS.credentialManager = manager + config.MTLS.Disabled = !connection.MTLSRequired + if connection.MTLSRequired { + config.MTLS.tlsConfig, managerErr = manager.TLSConfig() + if managerErr != nil { + return nil, managerErr + } + } + manager.Start(ctx) + } _, httpClient, err := newRuntimeClient(connection.RuntimeURL, config.AgentToken, config.MTLS) if err != nil { return nil, err diff --git a/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker.go b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker.go index d13573d..bcd33e9 100644 --- a/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker.go +++ b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker.go @@ -46,8 +46,9 @@ type RuntimeWorker struct { Logger *log.Logger OnReady func(RuntimeReadyPayload) - runtimeClient RuntimeClient - runtimeDialer RuntimeTransportDialer + runtimeClient RuntimeClient + runtimeDialer RuntimeTransportDialer + credentialManager *runtimeCredentialManager lifecycleMu sync.Mutex started bool @@ -126,8 +127,10 @@ func (node *RuntimeWorker) Start(parent context.Context) (retErr error) { if err := node.applyDefaultsAndValidate(); err != nil { return err } + connection := runtimeConnectionInformation{MTLSRequired: true} if node.runtimeClient == nil { - connection, err := resolveRuntimeConnection(parent, node.PlatformURL, node.RuntimeURL) + var err error + connection, err = resolveRuntimeConnection(parent, node.PlatformURL, node.RuntimeURL) if err != nil { return err } @@ -157,12 +160,46 @@ func (node *RuntimeWorker) Start(parent context.Context) (retErr error) { node.store = node.Store if node.runtimeClient == nil { + explicitMTLS := node.MTLS.CertFile != "" && node.MTLS.KeyFile != "" && node.MTLS.CAFile != "" + if !explicitMTLS || !connection.MTLSRequired { + credentialEndpoint := connection.CredentialEndpoint + if credentialEndpoint == "" && node.PlatformURL != "" { + platformOrigin, platformErr := validatePlatformOrigin(node.PlatformURL) + if platformErr != nil { + return platformErr + } + credentialEndpoint = platformOrigin + "/api/v1/runtime-credentials" + } + manager, managerErr := newRuntimeCredentialManager( + node.DataDir, credentialEndpoint, node.AgentToken, node.NodeID, node.AgentID, + node.NodeVersion, node.Capacity, node.Logger, + ) + if managerErr != nil { + return managerErr + } + if managerErr = manager.Ensure(startupCtx, false); managerErr != nil { + return managerErr + } + node.NodeID, node.AgentID = manager.Identity() + node.credentialManager = manager + node.MTLS.credentialManager = manager + node.MTLS.Disabled = !connection.MTLSRequired + if connection.MTLSRequired { + node.MTLS.tlsConfig, managerErr = manager.TLSConfig() + if managerErr != nil { + return managerErr + } + } + manager.Start(startupCtx) + } else { + node.MTLS.Disabled = false + } runtimeClient, httpClient, err := newRuntimeClient(node.RuntimeURL, node.AgentToken, node.MTLS) if err != nil { return err } node.runtimeClient = runtimeClient - node.runtimeDialer = &sdkRuntimeTransportDialer{runtime: runtimeClient} + node.runtimeDialer = &sdkRuntimeTransportDialer{runtime: runtimeClient, credentials: node.credentialManager} node.httpClient = httpClient } if node.runtimeDialer != nil { @@ -294,12 +331,19 @@ func (node *RuntimeWorker) applyDefaultsAndValidate() error { default: return errors.New("transport must be auto, ws, or pull") } - if !validRuntimeUUID(node.NodeID) { + explicitMTLS := node.MTLS.CertFile != "" || node.MTLS.KeyFile != "" || node.MTLS.CAFile != "" + if explicitMTLS && (node.MTLS.CertFile == "" || node.MTLS.KeyFile == "" || node.MTLS.CAFile == "") { + return errors.New("runtime mTLS cert, key, and CA files must be configured together") + } + if node.NodeID != "" && !validRuntimeUUID(node.NodeID) { return errors.New("RuntimeWorker ID must be a non-zero lowercase UUID") } - if !validRuntimeUUID(node.AgentID) { + if node.AgentID != "" && !validRuntimeUUID(node.AgentID) { return errors.New("Agent ID must be a non-zero lowercase UUID") } + if explicitMTLS && (!validRuntimeUUID(node.NodeID) || !validRuntimeUUID(node.AgentID)) { + return errors.New("RuntimeWorker ID and Agent ID are required with explicit mTLS files") + } if node.AgentToken == "" && node.runtimeClient == nil { return errors.New("Agent Token is required") } @@ -309,9 +353,6 @@ func (node *RuntimeWorker) applyDefaultsAndValidate() error { if node.Handler == nil { return errors.New("runtime handler is required") } - if node.runtimeClient == nil && (node.MTLS.CertFile == "" || node.MTLS.KeyFile == "" || node.MTLS.CAFile == "") { - return errors.New("runtime mTLS cert, key, and CA files are required") - } if node.Capacity == 0 { node.Capacity = RuntimeWorkerDefaultCapacity } @@ -410,7 +451,7 @@ func (node *RuntimeWorker) shutdown(ctx context.Context) error { node.store = nil } if node.httpClient != nil { - if transport, ok := node.httpClient.Transport.(*http.Transport); ok { + if transport, ok := node.httpClient.Transport.(interface{ CloseIdleConnections() }); ok { transport.CloseIdleConnections() } } diff --git a/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_client.go b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_client.go index bca5032..b2e4b22 100644 --- a/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_client.go +++ b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_client.go @@ -42,43 +42,62 @@ type RuntimeMTLSConfig struct { KeyFile string CAFile string ServerName string + // Disabled is set automatically from discovery when Core explicitly uses + // Agent Token-only HTTPS transport. + Disabled bool + tlsConfig *tls.Config + credentialManager *runtimeCredentialManager } func newRuntimeClient(runtimeAddress, agentToken string, config RuntimeMTLSConfig) (*Runtime, *http.Client, error) { - runtimeURL, err := validateRuntimeOrigin(runtimeAddress) + runtimeURL, err := validateRuntimeOriginForPolicy(runtimeAddress, config.Disabled) if err != nil { return nil, nil, err } if strings.TrimSpace(agentToken) == "" { return nil, nil, errors.New("Agent Token is required") } - if config.CertFile == "" || config.KeyFile == "" || config.CAFile == "" { - return nil, nil, errors.New("runtime mTLS cert, key, and CA files are required") - } - certificate, err := tls.LoadX509KeyPair(config.CertFile, config.KeyFile) - if err != nil { - return nil, nil, fmt.Errorf("load runtime mTLS client certificate: %w", err) - } - caPEM, err := os.ReadFile(config.CAFile) - if err != nil { - return nil, nil, fmt.Errorf("read runtime mTLS CA: %w", err) - } - roots := x509.NewCertPool() - if !roots.AppendCertsFromPEM(caPEM) { - return nil, nil, errors.New("runtime mTLS CA file contains no certificates") - } transport := http.DefaultTransport.(*http.Transport).Clone() - transport.TLSClientConfig = &tls.Config{ - MinVersion: tls.VersionTLS13, - Certificates: []tls.Certificate{certificate}, - RootCAs: roots, - ServerName: strings.TrimSpace(config.ServerName), + if !config.Disabled { + var tlsConfig *tls.Config + if config.tlsConfig != nil { + tlsConfig = config.tlsConfig.Clone() + } else { + if config.CertFile == "" || config.KeyFile == "" || config.CAFile == "" { + return nil, nil, errors.New("runtime mTLS credential is unavailable") + } + certificate, loadErr := tls.LoadX509KeyPair(config.CertFile, config.KeyFile) + if loadErr != nil { + return nil, nil, fmt.Errorf("load runtime mTLS client certificate: %w", loadErr) + } + caPEM, readErr := os.ReadFile(config.CAFile) + if readErr != nil { + return nil, nil, fmt.Errorf("read runtime mTLS CA: %w", readErr) + } + roots := x509.NewCertPool() + if !roots.AppendCertsFromPEM(caPEM) { + return nil, nil, errors.New("runtime mTLS CA file contains no certificates") + } + tlsConfig = &tls.Config{Certificates: []tls.Certificate{certificate}, RootCAs: roots} + } + tlsConfig.MinVersion = tls.VersionTLS13 + if serverName := strings.TrimSpace(config.ServerName); serverName != "" { + tlsConfig.ServerName = serverName + } + transport.TLSClientConfig = tlsConfig + } else { + transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12} } transport.ResponseHeaderTimeout = 35 * time.Second transport.TLSHandshakeTimeout = 10 * time.Second transport.IdleConnTimeout = 90 * time.Second + var roundTripper http.RoundTripper = transport + if config.credentialManager != nil { + config.credentialManager.SetCloseIdleConnections(transport.CloseIdleConnections) + roundTripper = &runtimeCredentialRenewingTransport{base: transport, credentials: config.credentialManager} + } httpClient := &http.Client{ - Transport: transport, + Transport: roundTripper, CheckRedirect: func(*http.Request, []*http.Request) error { // Runtime credentials and the client certificate are bound to the // configured Core origin. Runtime endpoints must not redirect them. @@ -97,3 +116,57 @@ func newRuntimeClient(runtimeAddress, agentToken string, config RuntimeMTLSConfi } return runtimeClient, httpClient, nil } + +type runtimeCredentialRenewingTransport struct { + base http.RoundTripper + credentials *runtimeCredentialManager +} + +func (transport *runtimeCredentialRenewingTransport) CloseIdleConnections() { + if closer, ok := transport.base.(interface{ CloseIdleConnections() }); ok { + closer.CloseIdleConnections() + } +} + +func (transport *runtimeCredentialRenewingTransport) RoundTrip(request *http.Request) (*http.Response, error) { + if transport == nil || transport.base == nil || transport.credentials == nil { + return nil, errors.New("runtime credential transport is unavailable") + } + if err := transport.credentials.Ensure(request.Context(), false); err != nil { + return nil, err + } + response, err := transport.base.RoundTrip(request) + if err == nil { + return response, nil + } + if !runtimeCredentialTLSFailure(err) { + return nil, err + } + if request.GetBody == nil && request.Body != nil { + return nil, err + } + if renewErr := transport.credentials.Ensure(request.Context(), true); renewErr != nil { + return nil, errors.Join(err, renewErr) + } + retry := request.Clone(request.Context()) + if request.GetBody != nil { + retry.Body, err = request.GetBody() + if err != nil { + return nil, err + } + } + return transport.base.RoundTrip(retry) +} + +func runtimeCredentialTLSFailure(err error) bool { + if err == nil { + return false + } + message := strings.ToLower(err.Error()) + for _, marker := range []string{"tls", "x509", "certificate", "unknown authority"} { + if strings.Contains(message, marker) { + return true + } + } + return false +} diff --git a/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_credentials.go b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_credentials.go new file mode 100644 index 0000000..ee43467 --- /dev/null +++ b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_credentials.go @@ -0,0 +1,487 @@ +package openlinker + +import ( + "bytes" + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "encoding/hex" + "encoding/json" + "encoding/pem" + "errors" + "fmt" + "io" + "log" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "sync" + "time" +) + +const ( + runtimeCredentialVersion = 1 + runtimeCredentialFile = "runtime-credential.json" + runtimeCredentialResponseMax = 64 << 10 + runtimeCredentialHTTPTimeout = 15 * time.Second + runtimeCredentialRetryAfter = 5 * time.Minute + runtimeCredentialExpiryMargin = 5 * time.Minute +) + +type runtimeCredentialDisk struct { + Version int `json:"version"` + NodeID string `json:"node_id"` + AgentID string `json:"agent_id,omitempty"` + PrivateKeyPEM string `json:"private_key_pem"` + CertificateChainPEM string `json:"certificate_chain_pem,omitempty"` + TrustBundlePEM string `json:"trust_bundle_pem,omitempty"` + CertificateSerial string `json:"certificate_serial,omitempty"` + PublicKeyThumbprint string `json:"public_key_thumbprint"` + CertificateNotBefore time.Time `json:"certificate_not_before,omitempty"` + CertificateNotAfter time.Time `json:"certificate_not_after,omitempty"` + RenewAfter time.Time `json:"renew_after,omitempty"` + Checksum string `json:"checksum"` +} + +type runtimeCredentialIssueRequest struct { + NodeID string `json:"node_id"` + DisplayName string `json:"display_name"` + NodeVersion string `json:"node_version"` + ProtocolVersion int `json:"protocol_version"` + RuntimeContractID string `json:"runtime_contract_id"` + RuntimeContractDigest string `json:"runtime_contract_digest"` + Features []string `json:"features"` + Capacity int64 `json:"capacity"` + CSRPEM string `json:"csr_pem"` +} + +type runtimeCredentialIssueResponse struct { + NodeID string `json:"node_id"` + AgentID string `json:"agent_id"` + CertificatePEM string `json:"certificate_pem"` + CertificateChainPEM string `json:"certificate_chain_pem"` + TrustBundlePEM string `json:"trust_bundle_pem"` + CertificateSerial string `json:"certificate_serial"` + PublicKeyThumbprint string `json:"public_key_thumbprint"` + NotBefore time.Time `json:"not_before"` + NotAfter time.Time `json:"not_after"` + RenewAfter time.Time `json:"renew_after"` + CertificateLifetimeHrs int `json:"certificate_lifetime_hours"` +} + +type runtimeCredentialManager struct { + mu sync.RWMutex + renewMu sync.Mutex + dataDir string + credentialEndpoint string + agentToken string + nodeVersion string + capacity int64 + logger *log.Logger + disk runtimeCredentialDisk + certificate tls.Certificate + rootCAs *x509.CertPool + closeIdle func() +} + +func newRuntimeCredentialManager( + dataDir, credentialEndpoint, agentToken, configuredNodeID, configuredAgentID, nodeVersion string, + capacity int64, + logger *log.Logger, +) (*runtimeCredentialManager, error) { + endpoint, err := validateRuntimeCredentialEndpoint(credentialEndpoint) + if err != nil { + return nil, err + } + absDir, err := filepath.Abs(dataDir) + if err != nil { + return nil, fmt.Errorf("resolve runtime credential directory: %w", err) + } + if err = ensurePrivateDataDir(absDir); err != nil { + return nil, err + } + disk, err := loadOrCreateRuntimeCredential(absDir, strings.TrimSpace(configuredNodeID)) + if err != nil { + return nil, err + } + if configuredNodeID != "" && disk.NodeID != configuredNodeID { + return nil, errors.New("configured RuntimeWorker ID differs from the key bound to this data directory") + } + if configuredAgentID != "" && disk.AgentID != "" && disk.AgentID != configuredAgentID { + return nil, errors.New("configured Agent ID differs from the credential bound to this data directory") + } + manager := &runtimeCredentialManager{ + dataDir: absDir, + credentialEndpoint: endpoint, + agentToken: strings.TrimSpace(agentToken), + nodeVersion: strings.TrimSpace(nodeVersion), + capacity: capacity, + logger: logger, + disk: disk, + } + if disk.CertificateChainPEM != "" { + if err = manager.loadTLSStateLocked(); err != nil { + return nil, err + } + } + return manager, nil +} + +func (m *runtimeCredentialManager) Ensure(ctx context.Context, force bool) error { + if m == nil { + return errors.New("runtime credential manager is unavailable") + } + m.renewMu.Lock() + defer m.renewMu.Unlock() + m.mu.RLock() + needsIssue := force || len(m.certificate.Certificate) == 0 || m.disk.CertificateNotAfter.IsZero() || + time.Now().Add(runtimeCredentialExpiryMargin).After(m.disk.CertificateNotAfter) || + (!m.disk.RenewAfter.IsZero() && !time.Now().Before(m.disk.RenewAfter)) + m.mu.RUnlock() + if !needsIssue { + return nil + } + return m.issue(ctx) +} + +func (m *runtimeCredentialManager) Start(ctx context.Context) { + if m == nil { + return + } + go func() { + for { + m.mu.RLock() + renewAfter := m.disk.RenewAfter + m.mu.RUnlock() + wait := time.Until(renewAfter) + if renewAfter.IsZero() || wait < time.Second { + wait = time.Second + } + timer := time.NewTimer(wait) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + renewCtx, cancel := context.WithTimeout(ctx, runtimeCredentialHTTPTimeout) + err := m.Ensure(renewCtx, false) + cancel() + if err != nil { + if m.logger != nil { + m.logger.Printf("openlinker: Runtime certificate renewal failed; retrying in %s: %v", runtimeCredentialRetryAfter, err) + } + timer = time.NewTimer(runtimeCredentialRetryAfter) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + } + } + }() +} + +func (m *runtimeCredentialManager) TLSConfig() (*tls.Config, error) { + m.mu.RLock() + defer m.mu.RUnlock() + if len(m.certificate.Certificate) == 0 || m.rootCAs == nil { + return nil, errors.New("runtime mTLS credential is unavailable") + } + return &tls.Config{ + MinVersion: tls.VersionTLS13, + RootCAs: m.rootCAs, + GetClientCertificate: func(*tls.CertificateRequestInfo) (*tls.Certificate, error) { + m.mu.RLock() + defer m.mu.RUnlock() + if len(m.certificate.Certificate) == 0 { + return nil, errors.New("runtime client certificate is unavailable") + } + certificate := m.certificate + return &certificate, nil + }, + }, nil +} + +func (m *runtimeCredentialManager) Identity() (string, string) { + m.mu.RLock() + defer m.mu.RUnlock() + return m.disk.NodeID, m.disk.AgentID +} + +func (m *runtimeCredentialManager) SetCloseIdleConnections(closeIdle func()) { + m.mu.Lock() + m.closeIdle = closeIdle + m.mu.Unlock() +} + +func (m *runtimeCredentialManager) issue(ctx context.Context) error { + m.mu.RLock() + disk := m.disk + m.mu.RUnlock() + key, err := parseRuntimeCredentialPrivateKey(disk.PrivateKeyPEM) + if err != nil { + return err + } + csrDER, err := x509.CreateCertificateRequest(rand.Reader, &x509.CertificateRequest{}, key) + if err != nil { + return fmt.Errorf("create Runtime certificate request: %w", err) + } + requestBody, err := json.Marshal(runtimeCredentialIssueRequest{ + NodeID: disk.NodeID, + DisplayName: "runtime-" + strings.ReplaceAll(disk.NodeID, "-", "")[:12], + NodeVersion: m.nodeVersion, + ProtocolVersion: RuntimeProtocolVersion, + RuntimeContractID: RuntimeContractID, + RuntimeContractDigest: RuntimeContractDigest, + Features: RuntimeRequiredFeatures(), + Capacity: m.capacity, + CSRPEM: string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE REQUEST", Bytes: csrDER})), + }) + if err != nil { + return err + } + request, err := http.NewRequestWithContext(ctx, http.MethodPost, m.credentialEndpoint, bytes.NewReader(requestBody)) + if err != nil { + return err + } + request.Header.Set("Authorization", "Bearer "+m.agentToken) + request.Header.Set("Content-Type", "application/json") + request.Header.Set("Accept", "application/json") + request.Header.Set("User-Agent", runtimeWorkerSDKAgent) + client := newRuntimeCredentialHTTPClient() + response, err := client.Do(request) + if err != nil { + return fmt.Errorf("request Runtime certificate: %w", err) + } + defer response.Body.Close() + body, err := io.ReadAll(io.LimitReader(response.Body, runtimeCredentialResponseMax+1)) + if err != nil { + return fmt.Errorf("read Runtime certificate response: %w", err) + } + if len(body) > runtimeCredentialResponseMax { + return errors.New("Runtime certificate response exceeds 64 KiB") + } + if response.StatusCode != http.StatusOK { + return fmt.Errorf("Runtime certificate request failed with HTTP %d", response.StatusCode) + } + var issued runtimeCredentialIssueResponse + decoder := json.NewDecoder(bytes.NewReader(body)) + decoder.DisallowUnknownFields() + if err = decoder.Decode(&issued); err != nil { + return fmt.Errorf("decode Runtime certificate response: %w", err) + } + if err = validateIssuedRuntimeCredential(disk, issued); err != nil { + return err + } + disk.AgentID = issued.AgentID + disk.CertificateChainPEM = issued.CertificateChainPEM + disk.TrustBundlePEM = issued.TrustBundlePEM + disk.CertificateSerial = strings.ToLower(issued.CertificateSerial) + disk.CertificateNotBefore = issued.NotBefore + disk.CertificateNotAfter = issued.NotAfter + disk.RenewAfter = issued.RenewAfter + if err = persistRuntimeCredential(m.dataDir, disk); err != nil { + return err + } + m.mu.Lock() + m.disk = disk + err = m.loadTLSStateLocked() + closeIdle := m.closeIdle + m.mu.Unlock() + if err != nil { + return err + } + if closeIdle != nil { + closeIdle() + } + return nil +} + +func (m *runtimeCredentialManager) loadTLSStateLocked() error { + certificate, err := tls.X509KeyPair([]byte(m.disk.CertificateChainPEM), []byte(m.disk.PrivateKeyPEM)) + if err != nil { + return fmt.Errorf("load automatic Runtime client certificate: %w", err) + } + if len(certificate.Certificate) == 0 { + return errors.New("automatic Runtime certificate chain is empty") + } + leaf, err := x509.ParseCertificate(certificate.Certificate[0]) + if err != nil { + return err + } + lifetime := leaf.NotAfter.Sub(leaf.NotBefore) + if leaf.NotAfter.Before(time.Now()) || lifetime < 23*time.Hour+50*time.Minute || + lifetime > 24*time.Hour+10*time.Minute { + return errors.New("automatic Runtime certificate validity is invalid") + } + roots := x509.NewCertPool() + if !roots.AppendCertsFromPEM([]byte(m.disk.TrustBundlePEM)) { + return errors.New("automatic Runtime trust bundle is invalid") + } + m.certificate = certificate + m.rootCAs = roots + return nil +} + +func loadOrCreateRuntimeCredential(dataDir, configuredNodeID string) (runtimeCredentialDisk, error) { + path := filepath.Join(dataDir, runtimeCredentialFile) + info, err := os.Lstat(path) + if errors.Is(err, os.ErrNotExist) { + nodeID := configuredNodeID + if nodeID == "" { + nodeID, err = newRuntimeUUID() + if err != nil { + return runtimeCredentialDisk{}, err + } + } + if !validRuntimeUUID(nodeID) { + return runtimeCredentialDisk{}, errors.New("RuntimeWorker ID must be a non-zero lowercase UUID") + } + key, keyErr := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if keyErr != nil { + return runtimeCredentialDisk{}, keyErr + } + keyDER, keyErr := x509.MarshalPKCS8PrivateKey(key) + if keyErr != nil { + return runtimeCredentialDisk{}, keyErr + } + spki, keyErr := x509.MarshalPKIXPublicKey(&key.PublicKey) + if keyErr != nil { + return runtimeCredentialDisk{}, keyErr + } + thumbprint := sha256.Sum256(spki) + disk := runtimeCredentialDisk{ + Version: runtimeCredentialVersion, + NodeID: nodeID, + PrivateKeyPEM: string(pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER})), + PublicKeyThumbprint: hex.EncodeToString(thumbprint[:]), + } + if err = persistRuntimeCredential(dataDir, disk); err != nil { + return runtimeCredentialDisk{}, err + } + return disk, nil + } + if err != nil { + return runtimeCredentialDisk{}, fmt.Errorf("inspect Runtime credential: %w", err) + } + if !info.Mode().IsRegular() || !runtimeFileModeIsPrivate(info.Mode()) || info.Size() <= 0 || info.Size() > runtimeCredentialResponseMax { + return runtimeCredentialDisk{}, errors.New("Runtime credential file is corrupt or not private") + } + raw, err := os.ReadFile(path) + if err != nil { + return runtimeCredentialDisk{}, err + } + var disk runtimeCredentialDisk + if err = decodeStrictJSON(raw, &disk); err != nil || disk.Version != runtimeCredentialVersion || + !validRuntimeUUID(disk.NodeID) || !runtimeCredentialChecksumValid(disk) { + return runtimeCredentialDisk{}, errors.New("Runtime credential file is corrupt") + } + key, err := parseRuntimeCredentialPrivateKey(disk.PrivateKeyPEM) + if err != nil { + return runtimeCredentialDisk{}, err + } + spki, err := x509.MarshalPKIXPublicKey(&key.PublicKey) + if err != nil { + return runtimeCredentialDisk{}, err + } + thumbprint := sha256.Sum256(spki) + if subtleString(disk.PublicKeyThumbprint) != hex.EncodeToString(thumbprint[:]) { + return runtimeCredentialDisk{}, errors.New("Runtime credential public key does not match its identity") + } + return disk, nil +} + +func persistRuntimeCredential(dataDir string, disk runtimeCredentialDisk) error { + disk.Version = runtimeCredentialVersion + disk.Checksum = "" + rawForChecksum, err := json.Marshal(disk) + if err != nil { + return err + } + digest := sha256.Sum256(rawForChecksum) + disk.Checksum = hex.EncodeToString(digest[:]) + raw, err := json.Marshal(disk) + if err != nil { + return err + } + if err = atomicWriteDurable(filepath.Join(dataDir, runtimeCredentialFile), raw, 0o600, nil); err != nil { + return fmt.Errorf("persist Runtime credential: %w", err) + } + return nil +} + +func runtimeCredentialChecksumValid(disk runtimeCredentialDisk) bool { + want := disk.Checksum + disk.Checksum = "" + raw, err := json.Marshal(disk) + if err != nil { + return false + } + digest := sha256.Sum256(raw) + return constantChecksumEqual(want, hex.EncodeToString(digest[:])) +} + +func parseRuntimeCredentialPrivateKey(value string) (*ecdsa.PrivateKey, error) { + block, rest := pem.Decode([]byte(value)) + if block == nil || block.Type != "PRIVATE KEY" || len(strings.TrimSpace(string(rest))) != 0 { + return nil, errors.New("Runtime credential private key is invalid") + } + parsed, err := x509.ParsePKCS8PrivateKey(block.Bytes) + if err != nil { + return nil, errors.New("Runtime credential private key is invalid") + } + key, ok := parsed.(*ecdsa.PrivateKey) + if !ok || key.Curve != elliptic.P256() { + return nil, errors.New("Runtime credential private key must use P-256") + } + return key, nil +} + +func validateIssuedRuntimeCredential(current runtimeCredentialDisk, issued runtimeCredentialIssueResponse) error { + if issued.NodeID != current.NodeID || !validRuntimeUUID(issued.AgentID) || + issued.CertificateChainPEM == "" || issued.TrustBundlePEM == "" || + issued.CertificateSerial == "" || issued.PublicKeyThumbprint != current.PublicKeyThumbprint || + issued.NotBefore.IsZero() || issued.NotAfter.IsZero() || !issued.NotBefore.Before(issued.NotAfter) || + issued.NotAfter.Sub(issued.NotBefore) < 23*time.Hour+50*time.Minute || + issued.NotAfter.Sub(issued.NotBefore) > 24*time.Hour+10*time.Minute || + issued.RenewAfter.Before(issued.NotBefore) || !issued.RenewAfter.Before(issued.NotAfter) { + return errors.New("Runtime certificate response is invalid") + } + return nil +} + +func validateRuntimeCredentialEndpoint(raw string) (string, error) { + value := strings.TrimSpace(raw) + parsed, err := url.Parse(value) + if err != nil || parsed.Host == "" || parsed.Hostname() == "" || parsed.User != nil || parsed.Fragment != "" { + return "", errors.New("Runtime credential endpoint is invalid") + } + if parsed.Scheme != "https" && (parsed.Scheme != "http" || !isLoopbackHost(parsed.Hostname())) { + return "", errors.New("Runtime credential endpoint must use HTTPS") + } + return parsed.String(), nil +} + +func newRuntimeCredentialHTTPClient() *http.Client { + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12} + return &http.Client{ + Transport: transport, + Timeout: runtimeCredentialHTTPTimeout, + CheckRedirect: func(*http.Request, []*http.Request) error { + return errors.New("Runtime credential endpoint redirects are not allowed") + }, + } +} + +func subtleString(value string) string { + return strings.ToLower(strings.TrimSpace(value)) +} diff --git a/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_discovery.go b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_discovery.go index 7798013..397f0c1 100644 --- a/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_discovery.go +++ b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_discovery.go @@ -27,11 +27,13 @@ type openLinkerDiscoveryManifest struct { Runtime string `json:"runtime"` } `json:"base_urls"` Runtime struct { - Enabled bool `json:"enabled"` - MTLSRequired bool `json:"mtls_required"` - Transports []string `json:"transports"` - DefaultTransport *string `json:"default_transport"` - TransportPolicy *openLinkerManifestTransportPolicy `json:"transport_policy"` + Enabled bool `json:"enabled"` + MTLSRequired bool `json:"mtls_required"` + CredentialEndpoint string `json:"credential_endpoint"` + TrustBundleEndpoint string `json:"trust_bundle_endpoint"` + Transports []string `json:"transports"` + DefaultTransport *string `json:"default_transport"` + TransportPolicy *openLinkerManifestTransportPolicy `json:"transport_policy"` } `json:"runtime"` } @@ -46,14 +48,17 @@ type openLinkerManifestTransportPolicy struct { } type runtimeConnectionInformation struct { - RuntimeURL string - Policy runtimeTransportPolicy + RuntimeURL string + Policy runtimeTransportPolicy + MTLSRequired bool + CredentialEndpoint string + TrustBundleEndpoint string } func resolveRuntimeConnection(ctx context.Context, platformURL, override string) (runtimeConnectionInformation, error) { if strings.TrimSpace(override) != "" { runtimeURL, err := validateRuntimeOrigin(override) - return runtimeConnectionInformation{RuntimeURL: runtimeURL, Policy: legacyRuntimeTransportPolicy()}, err + return runtimeConnectionInformation{RuntimeURL: runtimeURL, Policy: legacyRuntimeTransportPolicy(), MTLSRequired: true}, err } platformOrigin, err := validatePlatformOrigin(platformURL) if err != nil { @@ -95,13 +100,10 @@ func resolveRuntimeConnection(ctx context.Context, platformURL, override string) if !manifest.Runtime.Enabled { return runtimeConnectionInformation{}, errors.New("this OpenLinker instance does not provide a Runtime connection address") } - if !manifest.Runtime.MTLSRequired { - return runtimeConnectionInformation{}, errors.New("OpenLinker connection information does not require the expected mTLS identity") - } if strings.TrimSpace(manifest.BaseURLs.Runtime) == "" { return runtimeConnectionInformation{}, errors.New("this OpenLinker instance does not provide a Runtime connection address") } - runtimeURL, err := validateRuntimeOrigin(manifest.BaseURLs.Runtime) + runtimeURL, err := validateRuntimeOriginForPolicy(manifest.BaseURLs.Runtime, !manifest.Runtime.MTLSRequired) if err != nil { return runtimeConnectionInformation{}, err } @@ -109,7 +111,17 @@ func resolveRuntimeConnection(ctx context.Context, platformURL, override string) if err != nil { return runtimeConnectionInformation{}, err } - return runtimeConnectionInformation{RuntimeURL: runtimeURL, Policy: policy}, nil + credentialEndpoint := strings.TrimSpace(manifest.Runtime.CredentialEndpoint) + if credentialEndpoint == "" { + credentialEndpoint = platformOrigin + "/api/v1/runtime-credentials" + } + return runtimeConnectionInformation{ + RuntimeURL: runtimeURL, + Policy: policy, + MTLSRequired: manifest.Runtime.MTLSRequired, + CredentialEndpoint: credentialEndpoint, + TrustBundleEndpoint: strings.TrimSpace(manifest.Runtime.TrustBundleEndpoint), + }, nil } func resolveRuntimeURL(ctx context.Context, platformURL, override string) (string, error) { @@ -139,6 +151,10 @@ func validateRuntimeOrigin(raw string) (string, error) { return validateOrigin(raw, false, "Runtime connection address") } +func validateRuntimeOriginForPolicy(raw string, allowLoopbackHTTP bool) (string, error) { + return validateOrigin(raw, allowLoopbackHTTP, "Runtime connection address") +} + func validateOrigin(raw string, allowLoopbackHTTP bool, label string) (string, error) { value := strings.TrimSpace(raw) parsed, err := url.Parse(value) diff --git a/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_transport.go b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_transport.go index 5aec27e..535bbf2 100644 --- a/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_transport.go +++ b/vendor/github.com/OpenLinker-ai/openlinker-go/runtime_worker_transport.go @@ -44,8 +44,9 @@ type RuntimeTransportDialer interface { } type sdkRuntimeTransportDialer struct { - mu sync.RWMutex - runtime *Runtime + mu sync.RWMutex + runtime *Runtime + credentials *runtimeCredentialManager } func (dialer *sdkRuntimeTransportDialer) setRuntime(runtime *Runtime) { @@ -68,6 +69,18 @@ func (dialer *sdkRuntimeTransportDialer) DialRuntimeWebSocket( if runtime == nil { return nil, errors.New("runtime WebSocket dialer is unavailable") } + if dialer.credentials != nil { + if err := dialer.credentials.Ensure(ctx, false); err != nil { + return nil, err + } + } + client, err := runtime.DialRuntimeWebSocket(ctx, hello) + if err == nil || dialer.credentials == nil || !runtimeCredentialTLSFailure(err) { + return client, err + } + if renewErr := dialer.credentials.Ensure(ctx, true); renewErr != nil { + return nil, errors.Join(err, renewErr) + } return runtime.DialRuntimeWebSocket(ctx, hello) } @@ -76,6 +89,18 @@ func (dialer *sdkRuntimeTransportDialer) ProbeRuntimeWebSocket(ctx context.Conte if runtime == nil { return errors.New("runtime WebSocket dialer is unavailable") } + if dialer.credentials != nil { + if err := dialer.credentials.Ensure(ctx, false); err != nil { + return err + } + } + err := runtime.ProbeRuntimeWebSocket(ctx) + if err == nil || dialer.credentials == nil || !runtimeCredentialTLSFailure(err) { + return err + } + if renewErr := dialer.credentials.Ensure(ctx, true); renewErr != nil { + return errors.Join(err, renewErr) + } return runtime.ProbeRuntimeWebSocket(ctx) }