LinkPort/internal/store/assets.go

package store

import (
	"database/sql"
	"errors"
	"fmt"
	"golang.org/x/crypto/ssh"
	"linkport/internal/model"
	"time"
)

func (s *Store) GetAsset(id string) (model.Asset, error) {
	var a model.Asset
	var blob []byte
	err := s.db.QueryRow("SELECT config FROM assets WHERE id=?", id).Scan(&blob)
	if err == nil {
		err = s.unseal(blob, "asset:"+id, &a)
	}
	a.Project = ""
	return a, err
}

func (s *Store) Assets() ([]model.Asset, error) {
	rows, err := s.db.Query("SELECT id,config FROM assets ORDER BY id")
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := []model.Asset{}
	for rows.Next() {
		var id string
		var blob []byte
		var a model.Asset
		if err = rows.Scan(&id, &blob); err != nil {
			return nil, err
		}
		if err = s.unseal(blob, "asset:"+id, &a); err != nil {
			return nil, err
		}
		a.Project = ""
		result = append(result, a)
	}
	return result, rows.Err()
}

func (s *Store) saveAsset(tx *sql.Tx, a model.Asset) error {
	return s.saveAssetChecked(tx, a, true)
}

// Only use after validating the complete final bundle, so paired assets can
// rotate together without comparing one new half against one old half.
func (s *Store) saveAssetFromBundle(tx *sql.Tx, a model.Asset) error {
	return s.saveAssetChecked(tx, a, false)
}

func (s *Store) saveAssetChecked(tx *sql.Tx, a model.Asset, checkDependents bool) error {
	a.Project = ""
	if err := a.Validate(); err != nil {
		return err
	}
	var oldBlob []byte
	oldErr := tx.QueryRow("SELECT config FROM assets WHERE id=?", a.ID).Scan(&oldBlob)
	if oldErr != nil && !errors.Is(oldErr, sql.ErrNoRows) {
		return oldErr
	}
	var old model.Asset
	if oldErr == nil {
		if err := s.unseal(oldBlob, "asset:"+a.ID, &old); err != nil {
			return err
		}
		if old.Type != a.Type {
			return errors.New("asset type cannot change")
		}
	}
	if checkDependents {
		if err := s.validateAssetDependents(tx, a); err != nil {
			return err
		}
	}
	if err := s.recordHistory(tx, "assets", a.ID); err != nil {
		return err
	}
	blob, err := s.seal(a, "asset:"+a.ID)
	if err != nil {
		return err
	}
	_, err = tx.Exec("INSERT INTO assets(id,config) VALUES(?,?) ON CONFLICT(id) DO UPDATE SET config=excluded.config", a.ID, blob)
	if err == nil && checkDependents && oldErr == nil && old.Content != a.Content && a.Type == "kubeconfig" {
		err = s.clearAssetTargetHostKeys(tx, a.ID)
	}
	return err
}

func (s *Store) SaveAsset(a model.Asset) error {
	old, oldErr := s.GetAsset(a.ID)
	if oldErr != nil && !errors.Is(oldErr, sql.ErrNoRows) {
		return oldErr
	}
	if oldErr != nil || old.Content != a.Content {
		a.UpdatedAt = time.Now().UTC().Format(time.RFC3339Nano)
	} else {
		a.UpdatedAt = old.UpdatedAt
	}
	tx, err := s.db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if err = s.saveAsset(tx, a); err != nil {
		return err
	}
	return tx.Commit()
}

// Also runs during import and revision restore, inside the same transaction.
func (s *Store) clearAssetTargetHostKeys(tx *sql.Tx, assetID string) error {
	rows, err := tx.Query("SELECT id,config FROM outbounds")
	if err != nil {
		return err
	}
	profiles := map[string]bool{}
	for rows.Next() {
		var id string
		var blob []byte
		var p model.OutboundProfile
		if err = rows.Scan(&id, &blob); err == nil {
			err = s.unseal(blob, "outbound:"+id, &p)
		}
		if err != nil {
			rows.Close()
			return err
		}
		if p.Outbound.KubeConfigAssetID == assetID {
			profiles[id] = true
		}
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return err
	}
	rows, err = tx.Query("SELECT id,config FROM targets")
	if err != nil {
		return err
	}
	var targets []model.Target
	for rows.Next() {
		var id string
		var blob []byte
		var target model.Target
		if err = rows.Scan(&id, &blob); err == nil {
			err = s.unseal(blob, "target:"+id, &target)
		}
		if err != nil {
			rows.Close()
			return err
		}
		if (profiles[target.OutboundID] || target.Outbound.KubeConfigAssetID == assetID) && target.SSH.TargetHostKey != "" {
			target.SSH.TargetHostKey = ""
			targets = append(targets, target)
		}
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return err
	}
	for _, target := range targets {
		if err = s.saveTarget(tx, target); err != nil {
			return err
		}
	}
	return nil
}

func (s *Store) AssetReferences(id string) ([]string, error) {
	profiles, err := s.Outbounds()
	if err != nil {
		return nil, err
	}
	ids := []string{}
	for _, p := range profiles {
		if p.Outbound.KubeConfigAssetID == id {
			ids = append(ids, p.ID)
		}
	}
	return ids, nil
}

func (s *Store) DeleteAsset(id string) error {
	refs, err := s.AssetReferenceDetails(id)
	if err != nil {
		return err
	}
	if len(refs.OutboundIDs)+len(refs.TargetIDs)+len(refs.GatewayIDs) > 0 {
		return errors.New("asset is referenced by configuration")
	}
	return s.deleteConfig("assets", id)
}

func (s *Store) assetLookup(id, kind string) (string, error) {
	a, err := s.GetAsset(id)
	if err != nil || a.Type != kind {
		return "", fmt.Errorf("%s asset not found or wrong type", kind)
	}
	return a.Content, nil
}

func assetMapLookup(assets []model.Asset) model.AssetLookup {
	byID := map[string]model.Asset{}
	for _, a := range assets {
		byID[a.ID] = a
	}
	return func(id, kind string) (string, error) {
		a, ok := byID[id]
		if !ok || a.Type != kind {
			return "", fmt.Errorf("%s asset not found or wrong type", kind)
		}
		return a.Content, nil
	}
}

func (s *Store) txAssetLookup(tx *sql.Tx) model.AssetLookup {
	return func(id, kind string) (string, error) {
		var blob []byte
		var a model.Asset
		if err := tx.QueryRow("SELECT config FROM assets WHERE id=?", id).Scan(&blob); err != nil {
			return "", fmt.Errorf("%s asset not found", kind)
		}
		if err := s.unseal(blob, "asset:"+id, &a); err != nil {
			return "", err
		}
		if a.Type != kind {
			return "", fmt.Errorf("asset must have type %s", kind)
		}
		return a.Content, nil
	}
}

func (s *Store) ResolveTargetAssets(t model.Target) (model.Target, error) {
	return model.ResolveTargetAssets(t, s.assetLookup)
}
func (s *Store) ResolveGatewayAssets(p model.Proxy) (model.Proxy, error) {
	return model.ResolveGatewayAssets(p, s.assetLookup)
}

func checkAssetAccountKeys(t model.Target) error {
	if !model.IsSSH(t.Type) {
		return nil
	}
	for _, a := range t.Accounts {
		if a.AuthMethod != "privatekey" || (a.PrivateKeyAssetID == "" && a.CertificateAssetID == "") {
			continue
		}
		var signer ssh.Signer
		var err error
		if a.Passphrase != "" {
			signer, err = ssh.ParsePrivateKeyWithPassphrase([]byte(a.PrivateKey), []byte(a.Passphrase))
		} else {
			signer, err = ssh.ParsePrivateKey([]byte(a.PrivateKey))
		}
		if err != nil {
			return errors.New("invalid SSH private key asset or account passphrase")
		}
		if _, err = model.SSHCertificateSigner(a.Certificate, signer); err != nil {
			return errors.New("SSH certificate asset does not match account private key")
		}
	}
	return nil
}

func (s *Store) validateAssetDependents(tx *sql.Tx, a model.Asset) error {
	rows, err := tx.Query("SELECT id,config FROM targets")
	if err != nil {
		return err
	}
	var targets []model.Target
	for rows.Next() {
		var id string
		var blob []byte
		var t model.Target
		if err = rows.Scan(&id, &blob); err == nil {
			err = s.unseal(blob, "target:"+id, &t)
		}
		if err != nil {
			rows.Close()
			return err
		}
		for _, account := range t.Accounts {
			if account.PrivateKeyAssetID == a.ID || account.CertificateAssetID == a.ID {
				targets = append(targets, t)
				break
			}
		}
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return err
	}
	var gateways []model.Proxy
	if a.Type == "kubeconfig" {
		// Shared proxies select their own context. Validate a replacement
		// before it can invalidate every target or forwarding gateway using it.
		rows, err = tx.Query("SELECT id,config FROM outbounds")
		if err != nil {
			return err
		}
		for rows.Next() {
			var id string
			var blob []byte
			var profile model.OutboundProfile
			if err = rows.Scan(&id, &blob); err == nil {
				err = s.unseal(blob, "outbound:"+id, &profile)
			}
			if err == nil && profile.Outbound.KubeConfigAssetID == a.ID {
				profile.Outbound.KubeConfigContent = a.Content
				err = profile.Outbound.ValidateKubernetes()
			}
			if err != nil {
				rows.Close()
				return err
			}
		}
		err = rows.Err()
		rows.Close()
		if err != nil {
			return err
		}
		rows, err = tx.Query("SELECT id,config FROM gateways")
		if err != nil {
			return err
		}
		for rows.Next() {
			var id string
			var blob []byte
			var gateway model.Proxy
			if err = rows.Scan(&id, &blob); err == nil {
				err = s.unseal(blob, "proxy:"+id, &gateway)
			}
			if err != nil {
				rows.Close()
				return err
			}
			if gateway.TCPForward != nil && gateway.TCPForward.KubeConfigAssetID == a.ID {
				gateways = append(gateways, gateway)
			}
		}
		err = rows.Err()
		rows.Close()
		if err != nil {
			return err
		}
	}
	base := s.txAssetLookup(tx)
	lookup := func(id, kind string) (string, error) {
		if id == a.ID {
			if a.Type != kind {
				return "", errors.New("asset type mismatch")
			}
			return a.Content, nil
		}
		return base(id, kind)
	}
	for _, t := range targets {
		t, err = model.ResolveTargetAssets(t, lookup)
		if err != nil {
			return err
		}
		if err = checkAssetAccountKeys(t); err != nil {
			return err
		}
	}
	for _, gateway := range gateways {
		gateway.Outbound.KubeConfigContent = a.Content
		if err = gateway.ValidateTCPForward(); err != nil {
			return err
		}
	}
	return nil
}

type AssetReferences struct {
	OutboundIDs []string `json:"outboundIds"`
	TargetIDs   []string `json:"targetIds"`
	GatewayIDs  []string `json:"gatewayIds"`
}

// Includes direct and transitive dependencies, without materializing secrets.
func (s *Store) AssetReferenceDetails(id string) (AssetReferences, error) {
	refs := AssetReferences{OutboundIDs: []string{}, TargetIDs: []string{}, GatewayIDs: []string{}}
	var err error
	refs.OutboundIDs, err = s.AssetReferences(id)
	if err != nil {
		return refs, err
	}
	profiles := map[string]bool{}
	for _, p := range refs.OutboundIDs {
		profiles[p] = true
	}
	targets := map[string]bool{}
	rows, err := s.db.Query("SELECT id,config FROM targets")
	if err != nil {
		return refs, err
	}
	for rows.Next() {
		var tid string
		var blob []byte
		var t model.Target
		if err = rows.Scan(&tid, &blob); err == nil {
			err = s.unseal(blob, "target:"+tid, &t)
		}
		if err != nil {
			rows.Close()
			return refs, err
		}
		used := t.TargetCAAssetID == id || t.Outbound.KubeConfigAssetID == id || profiles[t.OutboundID]
		for _, a := range t.Accounts {
			used = used || a.PrivateKeyAssetID == id || a.CertificateAssetID == id
		}
		if used {
			targets[tid] = true
			refs.TargetIDs = append(refs.TargetIDs, tid)
		}
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return refs, err
	}
	rows, err = s.db.Query("SELECT id,config FROM gateways")
	if err != nil {
		return refs, err
	}
	for rows.Next() {
		var gid string
		var blob []byte
		var p model.Proxy
		if err = rows.Scan(&gid, &blob); err == nil {
			err = s.unseal(blob, "proxy:"+gid, &p)
		}
		if err != nil {
			rows.Close()
			return refs, err
		}
		used := targets[p.TargetID] || (p.TCPForward != nil && (p.TCPForward.KubeConfigAssetID == id || profiles[p.TCPForward.OutboundID]))
		for _, a := range p.GatewayAccounts {
			used = used || targets[a.TargetID] || a.AuthorizedKeysAssetID == id
		}
		if used {
			refs.GatewayIDs = append(refs.GatewayIDs, gid)
		}
	}
	err = rows.Err()
	rows.Close()
	return refs, err
}