package gatewayapi
import (
"cmp"
"context"
"crypto/tls"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net"
"net/url"
"os"
"strings"
"time"
exposurev1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/exposure/v1alpha1"
"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
"github.com/johnandersen777/socialweb-computer-kcp/internal/clientcert"
"github.com/johnandersen777/socialweb-computer-kcp/internal/component"
contractnodeclaim "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/nodeclaim"
"github.com/johnandersen777/socialweb-computer-kcp/internal/gateway/custody"
"github.com/johnandersen777/socialweb-computer-kcp/internal/gatewayhttp"
"github.com/johnandersen777/socialweb-computer-kcp/internal/guestnet"
"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpclient"
"github.com/johnandersen777/socialweb-computer-kcp/internal/kcplibs/impl/kcpstore"
"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/strs"
"github.com/johnandersen777/socialweb-computer-kcp/internal/vault/openbao"
)
const Kind = "GatewayAPI"
const (
DefaultShutdownTimeout = 30 * time.Second
DefaultDirectoryTTL = 60 * time.Second
DefaultDirectoryKeys = 10_000
DefaultVMReadySec = 120
DefaultAtprotoMount = "atproto"
DefaultAtprotoRole = custody.DefaultRole
DefaultCustodyTimeout = custody.DefaultTimeout
DefaultVaultClient = "socialweb-computer-k8s"
DefaultRenewBelow = 30 * time.Minute
DefaultRenewIncrement = time.Hour
)
type Deps struct {
component.Instance
KCP component.KCPClients
TLS *tls.Config
NodeClaimAgent *kcpclient.ServiceAccountRef
NodeClaims contractnodeclaim.Binder
Logins *openbao.Logins
}
func New(spec v1alpha1.GatewayAPISpec, deps Deps) (*Service, error) {
s := &Service{spec: spec, deps: deps}
s.log = deps.Log
if s.log == nil {
s.log = slog.New(slog.DiscardHandler)
}
s.workspace = cmp.Or(deps.Self.Workspace, "root")
s.stateDir = deps.StateDir
if spec.Storage != nil && spec.Storage.DataDir != "" {
s.stateDir = spec.Storage.DataDir
}
s.publicOrigin = cmp.Or(spec.PublicOrigin, publicOriginFor(spec.Hostnames, spec.TLS))
var serviceAudiences []string
if spec.ServiceAuth != nil {
serviceAudiences = spec.ServiceAuth.Audience
}
s.audiences = uniqueStrings(append(gatewayAudiences(spec.Audience, serviceAudiences, spec.Identity),
gatewayHostAudiences(spec.Hostnames, spec.TLS, spec.PublicOrigin, spec.Network)...))
if nm := spec.NodeMarket; nm != nil {
args, err := parseNodePolicyArgs(nm.NodePolicyArgs)
if err != nil {
return nil, fmt.Errorf("gatewayapi: nodeMarket.nodePolicyArgs: %w", err)
}
s.policyArgs = args
s.homeWorkspaces = []string{s.workspace}
if nm.HomeWorkspace != "" {
s.homeWorkspaces = []string{nm.HomeWorkspace}
}
s.contractWorkspace = cmp.Or(nm.ContractWorkspace, s.workspace)
}
if spec.ClusterProvider != nil {
if err := checkClusterProvider(*spec.ClusterProvider); err != nil {
return nil, err
}
}
if err := s.check(); err != nil {
return nil, err
}
caFile := cmp.Or(spec.KubeconfigCAFile, os.Getenv("SOCIALWEB_GATEWAY_KCP_PUBLIC_CA_FILE"))
server, err := gatewayServerFor(deps.KCP, deps.NodeClaimAgent, s.workspace, kubeconfigIssuer(spec.ClusterStateDir, caFile, deps.Logins),
os.Getenv("SOCIALWEB_GATEWAY_KCP_PUBLIC_HOST"), spec.ClusterStateDir)
if err != nil {
return nil, err
}
s.server = server
return s, nil
}
func (s *Service) check() error {
if s.spec.NodesOnly {
return s.checkNodesOnly()
}
switch {
case len(s.audiences) == 0:
return errors.New("gatewayapi: audience, serviceAuth.audience, identity.did or a hostname is required")
case !s.vaultConfigured():
return errors.New("gatewayapi: a vault ref is required for sign-in")
}
if s.spec.Network != nil {
for _, host := range s.spec.Network.Hosts {
if host.Name == "" || host.Address == "" {
return errors.New("gatewayapi: every network host needs a name and an address")
}
}
}
if s.publicOrigin != "" {
if err := checkOrigin(s.publicOrigin); err != nil {
return fmt.Errorf("gatewayapi: PublicOrigin: %w", err)
}
}
if len(s.spec.RequiredDownloads) > 0 && s.downloadsDir() == "" {
return errors.New("gatewayapi: RequiredDownloads needs DownloadsDir")
}
if needsMeshd(s.spec.RequiredDownloads) && s.spec.Network == nil && s.downloadBase() == "" {
return errors.New("gatewayapi: socialweb-meshd is required but no download base is set")
}
if s.spec.Serve != nil {
if err := checkLoopbackHost(s.spec.Serve.Host); err != nil {
return err
}
}
return nil
}
func (s *Service) checkNodesOnly() error {
switch {
case s.spec.NodeMarket == nil:
return errors.New("gatewayapi: nodesOnly needs a nodeMarket")
case strings.Trim(s.spec.NodeMarket.VaultMount, "/") == "":
return errors.New("gatewayapi: nodesOnly needs nodeMarket.vaultMount, the KV mount of the node join bundles")
case !s.vaultConfigured():
return errors.New("gatewayapi: nodesOnly needs the vault ref of the node join bundles")
}
if s.spec.PublicOrigin != "" {
if err := checkDownloadBase(s.spec.PublicOrigin); err != nil {
return fmt.Errorf("gatewayapi: nodesOnly publicOrigin, the base nodes fetch /downloads// from: %w", err)
}
}
return nil
}
func checkDownloadBase(raw string) error {
u, err := url.Parse(strings.TrimSpace(raw))
if err != nil {
return err
}
if u.Scheme != "https" || u.Host == "" {
return fmt.Errorf("%q is not an absolute https URL", raw)
}
if u.RawQuery != "" || u.Fragment != "" {
return fmt.Errorf("%q carries a query or fragment", raw)
}
return nil
}
func (s *Service) vaultConfigured() bool {
return s.spec.VaultRef != nil && s.spec.VaultRef.Name != ""
}
func (s *Service) downloadsDir() string {
if s.spec.Storage == nil {
return ""
}
return s.spec.Storage.BlobsDir
}
func (s *Service) clusterCADir() string {
if s.spec.Storage == nil {
return ""
}
return s.spec.Storage.CertsDir
}
func needsMeshd(required []string) bool {
for _, name := range required {
if strings.HasPrefix(name, gatewayhttp.MeshdDownloadPrefix) {
return true
}
}
return false
}
func (s *Service) downloadBase() string {
return strings.TrimRight(s.publicOrigin, "/")
}
func (s *Service) localDownloadBase(ctx context.Context) string {
workspace := s.workspace
store, err := kcpStoreFor(s.deps.KCP, workspace)
if err != nil {
return ""
}
services, err := kcpstore.Of[exposurev1alpha1.Service](store, exposureServiceGVR).List(ctx, workspace)
if err != nil {
return ""
}
serviceName := ""
for i := range services {
target := services[i].Spec.TargetRef
targetWorkspace := target.Workspace
targetWorkspace = cmp.Or(targetWorkspace, workspace)
if target.Kind == Kind && target.Name == s.deps.Self.Name && targetWorkspace == workspace {
serviceName = services[i].Name
break
}
}
if serviceName == "" {
return ""
}
exposures, err := kcpstore.Of[exposurev1alpha1.Exposure](store, exposureGVR).List(ctx, workspace)
if err != nil {
return ""
}
for i := range exposures {
exp := &exposures[i]
if exp.Spec.Mode != exposurev1alpha1.ExposureModeLocal {
continue
}
if strs.LastSegment(exp.Spec.ServiceRef) != serviceName {
continue
}
parsed, err := url.Parse(strings.TrimSpace(exp.Status.Address))
if err != nil || parsed.Port() == "" {
continue
}
return "http://" + guestnet.HostAddress + ":" + parsed.Port()
}
return ""
}
func checkOrigin(raw string) error {
u, err := url.Parse(strings.TrimSpace(raw))
if err != nil {
return err
}
if (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
return fmt.Errorf("%q is not an absolute http or https origin", raw)
}
if u.Scheme == "http" && !custody.LoopbackHost(u.Hostname()) {
return fmt.Errorf("%q is plain http on a non-loopback host, so the session cookie could not be Secure", raw)
}
if u.RawQuery != "" || u.Fragment != "" {
return fmt.Errorf("%q carries a query or fragment", raw)
}
if u.Path != "" && u.Path != "/" {
return fmt.Errorf("%q carries a path, not an origin", raw)
}
return nil
}
func checkLoopbackHost(host string) error {
if custody.LoopbackHost(host) {
return nil
}
return fmt.Errorf("gatewayapi: host %q is not loopback; this listener carries session cookies and service-auth tokens, so it binds 127.0.0.1 or [::1] only", host)
}
func kubeconfigIssuer(clusterStateDir, caFile string, logins *openbao.Logins) *clientcert.Issuer {
if clusterStateDir == "" {
return nil
}
return &clientcert.Issuer{StateDir: clusterStateDir, CAFile: caFile, Logins: logins}
}
func checkClusterProvider(cp v1alpha1.GatewayClusterProvider) error {
if raw := strings.TrimSpace(cp.ScaleHold); raw != "" {
hold, err := time.ParseDuration(raw)
if err != nil {
return fmt.Errorf("gatewayapi: clusterProvider.scaleHold %q is not a duration: %w", raw, err)
}
if hold < 0 {
return fmt.Errorf("gatewayapi: clusterProvider.scaleHold %q must not be negative", raw)
}
}
if cp.ScaleMaxNodes != 0 {
return errors.New("gatewayapi: clusterProvider.scaleMaxNodes belongs to the clusterprovider host process; the GatewayAPI component cannot honor it")
}
if cp.ScalePodsPerNode != 0 {
return errors.New("gatewayapi: clusterProvider.scalePodsPerNode belongs to the clusterprovider host process; the GatewayAPI component cannot honor it")
}
return nil
}
func parseNodePolicyArgs(raw string) (map[string]any, error) {
if strings.TrimSpace(raw) == "" {
return nil, nil
}
var value any
if err := json.Unmarshal([]byte(raw), &value); err != nil {
return nil, fmt.Errorf("not JSON: %w", err)
}
if value == nil {
return nil, nil
}
args, ok := value.(map[string]any)
if !ok {
return nil, errors.New("must be a JSON object")
}
return args, nil
}
func gatewayAudiences(audience string, audiences []string, identity *v1alpha1.ComponentIdentity) []string {
out := []string{}
if trimmed := strings.TrimSpace(audience); trimmed != "" {
out = append(out, trimmed)
}
for _, value := range audiences {
if trimmed := strings.TrimSpace(value); trimmed != "" {
out = append(out, trimmed)
}
}
if len(out) == 0 && identity != nil {
if did := strings.TrimSpace(identity.Did); did != "" {
out = append(out, did)
}
}
return uniqueStrings(out)
}
func gatewayHostAudiences(hostnames []string, tls *v1alpha1.ComponentTLS, publicOrigin string, network *v1alpha1.MarketNetworkValue) []string {
candidates := append([]string(nil), hostnames...)
if tls != nil {
candidates = append(candidates, tls.Hostnames...)
}
if publicOrigin != "" {
candidates = append(candidates, publicOrigin)
}
if network != nil {
candidates = append(candidates, network.IngressProxyHost)
}
out := []string{}
for _, raw := range candidates {
raw = strings.TrimSpace(raw)
if raw == "" {
continue
}
host := raw
if parsed, err := url.Parse(raw); err == nil && parsed.Host != "" {
host = parsed.Host
}
hosts := []string{host, strings.ToLower(host)}
if name, _, err := net.SplitHostPort(host); err == nil && name != "" {
hosts = append(hosts, name, strings.ToLower(name))
}
for _, value := range hosts {
if value == "" {
continue
}
out = append(out, "did:web:"+value)
}
}
return out
}
func uniqueStrings(values []string) []string {
seen := map[string]bool{}
out := make([]string, 0, len(values))
for _, value := range values {
if seen[value] {
continue
}
seen[value] = true
out = append(out, value)
}
return out
}
func publicOriginFor(hostnames []string, tls *v1alpha1.ComponentTLS) string {
if tls == nil || !tls.Enabled {
return ""
}
candidates := hostnames
if len(tls.Hostnames) > 0 {
candidates = tls.Hostnames
}
for _, hostname := range candidates {
hostname = strings.TrimSpace(hostname)
hostname = strings.TrimPrefix(hostname, "https://")
hostname = strings.TrimPrefix(hostname, "http://")
hostname = strings.TrimRight(hostname, "/")
if hostname == "" {
continue
}
return "https://" + hostname
}
return ""
}