gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-13 07:41:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1864 from openziti/fix.perms.on.controller.list

Browse Source 
Fix Controller List & Pub Key Sync
This commit is contained in:
Andrew
2024-03-26 11:53:30 -04:00
committed by GitHub
parent 27c19a52dd e59f9b8d66
commit 054bb2052a
26 changed files with 1542 additions and 973 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/pb/edge_cmd_pb/edge_cmd.pb.go
+1015 -855
View File
File diff suppressed because it is too large Load Diff
common/pb/edge_cmd_pb/edge_cmd.proto
+10
View File
@@ -175,6 +175,16 @@ message Controller {
bool isOnline = 6;
google.protobuf.Timestamp lastJoinedAt = 7;
map<string, TagValue> tags = 8;
map<string,ApiAddressList> apiAddresses = 9;
}
message ApiAddressList {
repeated ApiAddress addresses = 1;
}
message ApiAddress {
string Url = 1;
string Version = 2;
}
// Edge Routers
common/router_data_model.go
+4
View File
@@ -417,6 +417,10 @@ func (rdm *RouterDataModel) ApplyRevocationEvent(event *edge_ctrl_pb.DataState_E
rdm.sendEvent(event)
}
func (rdm *RouterDataModel) GetPublicKeys() map[string]*edge_ctrl_pb.DataState_PublicKey {
return rdm.PublicKeys.Items()
}
func (rdm *RouterDataModel) GetDataState() *edge_ctrl_pb.DataState {
var events []*edge_ctrl_pb.DataState_Event
controller/controller.go
+71 -1
View File
@@ -31,6 +31,7 @@ import (
"github.com/openziti/ziti/controller/zac"
"math/big"
"os"
"sync"
"sync/atomic"
"time"
@@ -85,6 +86,10 @@ type Controller struct {
metricsRegistry metrics.Registry
versionProvider versions.VersionProvider
eventDispatcher *events.Dispatcher
apiData map[string][]event.ApiAddress
apiDataBytes []byte
apiDataOnce sync.Once
}
func (c *Controller) GetPeerSigners() []*x509.Certificate {
@@ -154,6 +159,15 @@ func (c *Controller) GetRaftIndex() uint64 {
return c.raftController.Raft.LastIndex()
}
func (c *Controller) GetRaftInfo() (string, string, string) {
id := c.config.Id.Token
addr := c.raftController.Mesh.Addr().String()
version := c.GetVersionProvider().Version()
return string(addr), string(id), version
}
func (c *Controller) GetDb() boltz.Db {
return c.config.Db
}
@@ -338,7 +352,11 @@ func (c *Controller) Run() error {
panic(err)
}
go c.xweb.Run()
c.xweb.Run()
for _, helloHeader := range c.GetHelloHeaderProviders() {
helloHeader.Apply(headers)
}
// event handlers
if err := c.eventDispatcher.WireEventHandlers(c.getEventHandlerConfigs()); err != nil {
@@ -562,3 +580,55 @@ func (c *Controller) InitializeRaftFromBoltDb(sourceDbPath string) error {
return c.raftController.Dispatch(cmd)
}
// TODO: this functions is a temporary hack and should be provided by xweb
func getApiPath(binding string) string {
switch binding {
case "edge-client":
return "/edge/client/v1"
case "edge-management":
return "/edge/management/v1"
case "fabric":
return "/fabric/v1"
case "health-checks":
return "health-checks"
case "edge-oidc":
return "/oidc"
}
return ""
}
func (c *Controller) GetApiAddresses() (map[string][]event.ApiAddress, []byte) {
c.apiDataOnce.Do(func() {
xwebConfig := c.xweb.GetConfig()
apiData := map[string][]event.ApiAddress{}
for _, serverConfig := range xwebConfig.ServerConfigs {
for _, bindPoint := range serverConfig.BindPoints {
for _, api := range serverConfig.APIs {
apiData[api.Binding()] = append(apiData[api.Binding()], event.ApiAddress{
Url: "https://" + bindPoint.Address + getApiPath(api.Binding()), //TODO: temp till xweb support reporting API paths
Version: "v1", //TODO: temp till xweb supports reporting versions via api.Version()
})
}
}
}
c.apiData = apiData
c.apiDataBytes, _ = json.Marshal(apiData)
})
return c.apiData, c.apiDataBytes
}
func (c *Controller) GetHelloHeaderProviders() []mesh.HeaderProvider {
providerFunc := func(headers map[int32][]byte) {
_, apiDataBytes := c.GetApiAddresses()
headers[mesh.ApiAddressesHeader] = apiDataBytes
}
provider := mesh.HeaderProviderFunc(providerFunc)
return []mesh.HeaderProvider{provider}
}
controller/db/controller_store.go
+58 -9
View File
@@ -19,25 +19,35 @@ package db
import (
"github.com/openziti/storage/ast"
"github.com/openziti/storage/boltz"
"strconv"
"time"
)
const (
FieldControllerAddress = "address"
FieldControllerCertPem = "certPem"
FieldControllerFingerprint = "fingerprint"
FieldControllerIsOnline = "isOnline"
FieldControllerLastJoinedAt = "lastJoinedAt"
FieldControllerCtrlAddress = "ctrlAddress"
FieldControllerCertPem = "certPem"
FieldControllerFingerprint = "fingerprint"
FieldControllerIsOnline = "isOnline"
FieldControllerLastJoinedAt = "lastJoinedAt"
FieldControllerApiAddresses = "apiAddresses"
FieldControllerApiAddressVersion = "apiAddresses.version"
FieldControllerApiAddressUrl = "apiAddresses.url"
)
type Controller struct {
boltz.BaseExtEntity
Name string `json:"name"`
Address string `json:"address"`
CtrlAddress string `json:"address"`
CertPem string `json:"certPem"`
Fingerprint string `json:"fingerprint"`
IsOnline bool `json:"isOnline"`
LastJoinedAt *time.Time `json:"lastJoinedAt"`
ApiAddresses map[string][]ApiAddress
}
type ApiAddress struct {
Url string `json:"url"`
Version string `json:"version"`
}
func (entity *Controller) GetName() string {
@@ -75,7 +85,7 @@ func (store *controllerStoreImpl) initializeLocal() {
store.AddExtEntitySymbols()
store.indexName = store.addUniqueNameField()
store.AddSymbol(FieldControllerAddress, ast.NodeTypeString)
store.AddSymbol(FieldControllerCtrlAddress, ast.NodeTypeString)
store.AddSymbol(FieldControllerCertPem, ast.NodeTypeString)
store.AddSymbol(FieldControllerFingerprint, ast.NodeTypeString)
store.AddSymbol(FieldControllerIsOnline, ast.NodeTypeBool)
@@ -91,19 +101,58 @@ func (store *controllerStoreImpl) NewEntity() *Controller {
func (store *controllerStoreImpl) FillEntity(entity *Controller, bucket *boltz.TypedBucket) {
entity.LoadBaseValues(bucket)
entity.Name = bucket.GetStringOrError(FieldName)
entity.Address = bucket.GetStringWithDefault(FieldControllerAddress, "")
entity.CtrlAddress = bucket.GetStringWithDefault(FieldControllerCtrlAddress, "")
entity.CertPem = bucket.GetStringOrError(FieldControllerCertPem)
entity.Fingerprint = bucket.GetStringOrError(FieldControllerFingerprint)
entity.IsOnline = bucket.GetBoolWithDefault(FieldControllerIsOnline, false)
entity.LastJoinedAt = bucket.GetTime(FieldControllerLastJoinedAt)
entity.ApiAddresses = map[string][]ApiAddress{}
apiListBucket := bucket.GetBucket(FieldControllerApiAddresses)
if apiListBucket != nil {
apiListCursor := apiListBucket.Cursor()
for apiKey, _ := apiListCursor.First(); apiKey != nil; apiKey, _ = apiListCursor.Next() {
apiBucket := apiListBucket.GetBucket(string(apiKey))
if apiBucket != nil {
entity.ApiAddresses[string(apiKey)] = nil
instanceCursor := apiBucket.Cursor()
for instanceKey, _ := instanceCursor.First(); instanceKey != nil; instanceKey, _ = instanceCursor.Next() {
instance := apiBucket.GetBucket(string(instanceKey))
if instance != nil {
newInstance := ApiAddress{
Url: instance.GetStringWithDefault(FieldControllerApiAddressUrl, ""),
Version: instance.GetStringWithDefault(FieldControllerApiAddressVersion, ""),
}
entity.ApiAddresses[string(apiKey)] = append(entity.ApiAddresses[string(apiKey)], newInstance)
}
}
}
}
}
}
func (store *controllerStoreImpl) PersistEntity(entity *Controller, ctx *boltz.PersistContext) {
entity.SetBaseValues(ctx)
ctx.SetString(FieldName, entity.Name)
ctx.SetString(FieldControllerAddress, entity.Address)
ctx.SetString(FieldControllerCtrlAddress, entity.CtrlAddress)
ctx.SetString(FieldControllerCertPem, entity.CertPem)
ctx.SetString(FieldControllerFingerprint, entity.Fingerprint)
ctx.SetBool(FieldControllerIsOnline, entity.IsOnline)
ctx.SetTimeP(FieldControllerLastJoinedAt, entity.LastJoinedAt)
apiListBucket := ctx.Bucket.GetOrCreateBucket(FieldControllerApiAddresses)
for apiKey, apis := range entity.ApiAddresses {
apiBucket, _ := apiListBucket.EmptyBucket(apiKey)
for i, instance := range apis {
instanceBucket := apiBucket.GetOrCreateBucket(strconv.Itoa(i))
instanceBucket.SetString(FieldControllerApiAddressUrl, instance.Url, ctx.FieldChecker)
instanceBucket.SetString(FieldControllerApiAddressVersion, instance.Version, ctx.FieldChecker)
}
}
}
controller/env/appenv.go
Vendored
+12 -22
View File
@@ -314,6 +314,8 @@ type HostController interface {
GetEventDispatcher() event.Dispatcher
GetRaftIndex() uint64
GetPeerAddresses() []string
GetRaftInfo() (string, string, string)
GetApiAddresses() (map[string][]event.ApiAddress, []byte)
}
type Schemes struct {
@@ -830,29 +832,19 @@ func (ae *AppEnv) ControllersKeyFunc(token *jwt.Token) (interface{}, error) {
return nil, nil
}
return ae.GetControllerPublicKey(kid), nil
key := ae.GetControllerPublicKey(kid)
if key == nil {
return nil, fmt.Errorf("key for kid %s, not found", kid)
}
return key, nil
}
func (ae *AppEnv) GetControllerPublicKey(kid string) crypto.PublicKey {
serverTlsCert, serverKid, _ := ae.GetServerCert()
signers := ae.GetHostController().GetPeerSigners()
kids := map[string]crypto.PublicKey{
serverKid: serverTlsCert.Leaf.PublicKey,
}
for _, signer := range signers {
signerKid := fmt.Sprintf("%s", sha1.Sum(signer.Raw))
kids[signerKid] = signer.PublicKey
}
for curKid, publicKey := range kids {
if curKid == kid {
return publicKey
}
}
return nil
signers := ae.Broker.GetPublicKeys()
pfxlog.Logger().Info("looking for signer: " + kid)
return signers[kid]
}
func (ae *AppEnv) CreateRequestContext(rw http.ResponseWriter, r *http.Request) *response.RequestContext {
@@ -967,8 +959,6 @@ func (ae *AppEnv) SetServerCert(serverCert *tls.Certificate) {
signMethod := getJwtSigningMethod(serverCert)
kid := fmt.Sprintf("%x", sha1.Sum(serverCert.Certificate[0]))
ae.serverSigner = jwtsigner.New(signMethod, serverCert.PrivateKey, kid)
ae.Broker.routerSyncStrategy.AddPublicKey(ae.ServerCert)
}
func (ae *AppEnv) OidcIssuer() string {
controller/env/broker.go
Vendored
+51 -6
View File
@@ -17,6 +17,8 @@
package env
import (
"crypto"
"crypto/x509"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel/v2"
"github.com/openziti/storage/boltz"
@@ -25,6 +27,7 @@ import (
"github.com/openziti/ziti/controller/event"
"github.com/openziti/ziti/controller/network"
"go.etcd.io/bbolt"
"sync"
)
const (
@@ -57,6 +60,9 @@ type Broker struct {
apiSessionChunkSize int
routerMsgBufferSize int
routerSyncStrategy RouterSyncStrategy
publicKeyLock sync.Mutex
publicKeys map[string]crypto.PublicKey
}
func NewBroker(ae *AppEnv, synchronizer RouterSyncStrategy) *Broker {
@@ -73,13 +79,11 @@ func NewBroker(ae *AppEnv, synchronizer RouterSyncStrategy) *Broker {
broker.ae.GetStores().ApiSession.GetEventsEmitter().AddListener(db.EventFullyAuthenticated, broker.apiSessionFullyAuthenticated)
broker.ae.GetStores().ApiSessionCertificate.AddEntityEventListenerF(broker.apiSessionCertificateCreated, boltz.EntityCreatedAsync)
broker.ae.GetStores().ApiSessionCertificate.AddEntityEventListenerF(broker.apiSessionCertificateDeleted, boltz.EntityDeletedAsync)
broker.ae.GetStores().Controller.AddEntityEventListenerF(broker.controllerAdded, boltz.EntityCreated)
broker.ae.GetStores().Controller.AddEntityEventListenerF(broker.controllerAdded, boltz.EntityUpdated)
ae.HostController.GetNetwork().AddRouterPresenceHandler(broker)
ae.HostController.GetEventDispatcher().AddClusterEventHandler(broker)
ae.GetHostController().GetPeerSigners()
//updates controller store on leader, store update trigger strategy in all controllers
ae.HostController.GetEventDispatcher().AddClusterEventHandler(broker)
return broker
}
@@ -187,6 +191,47 @@ func (broker *Broker) Stop() {
broker.routerSyncStrategy.Stop()
}
func (broker *Broker) controllerAdded(controller *db.Controller) {
broker.routerSyncStrategy.PeerAdded(controller)
func (broker *Broker) GetPublicKeys() map[string]crypto.PublicKey {
broker.publicKeyLock.Lock()
defer broker.publicKeyLock.Unlock()
if broker.publicKeys == nil {
broker.publicKeys = map[string]crypto.PublicKey{}
}
for kid, pubKey := range broker.routerSyncStrategy.GetPublicKeys() {
//don't reprocess the same kid
if _, exists := broker.publicKeys[kid]; exists {
continue
}
log := pfxlog.Logger().WithField("format", pubKey.Format).WithField("kid", kid)
switch pubKey.Format {
case edge_ctrl_pb.DataState_PublicKey_X509CertDer:
cert, err := x509.ParseCertificate(pubKey.GetData())
if err != nil {
log.WithError(err).Error("error parsing x509 certificate DER")
continue
}
broker.publicKeys[kid] = cert.PublicKey
case edge_ctrl_pb.DataState_PublicKey_PKIXPublicKey:
pub, err := x509.ParsePKIXPublicKey(pubKey.GetData())
if err != nil {
log.WithError(err).Error("error parsing PKIX public key DER")
continue
}
broker.publicKeys[kid] = pub
default:
log.Error("unknown public key format")
}
}
result := map[string]crypto.PublicKey{}
for k, v := range broker.publicKeys {
result[k] = v
}
return result
}
controller/env/sync.go
Vendored
+2 -5
View File
@@ -17,9 +17,9 @@
package env
import (
"crypto/tls"
"github.com/openziti/channel/v2"
"github.com/openziti/foundation/v2/versions"
"github.com/openziti/ziti/common/pb/edge_ctrl_pb"
"github.com/openziti/ziti/controller/db"
"github.com/openziti/ziti/controller/model"
"github.com/openziti/ziti/controller/network"
@@ -59,9 +59,9 @@ type RouterSyncStrategy interface {
Type() RouterSyncStrategyType
GetEdgeRouterState(id string) RouterStateValues
Stop()
GetPublicKeys() map[string]*edge_ctrl_pb.DataState_PublicKey
RouterConnectionHandler
RouterSynchronizerEventHandler
AddPublicKey(cert *tls.Certificate)
}
// RouterConnectionHandler is responsible for handling router connect/disconnect for synchronizing state.
@@ -78,10 +78,7 @@ type RouterSynchronizerEventHandler interface {
ApiSessionAdded(apiSession *db.ApiSession)
ApiSessionUpdated(apiSession *db.ApiSession, apiSessionCert *db.ApiSessionCertificate)
ApiSessionDeleted(apiSession *db.ApiSession)
SessionDeleted(session *db.Session)
PeerAdded(*db.Controller)
}
// RouterState provides a thread save mechanism to access and set router status information that may be influx
controller/event/cluster.go
+10 -4
View File
@@ -37,10 +37,16 @@ const (
)
type ClusterPeer struct {
Id string `json:"id,omitempty"`
Addr string `json:"addr,omitempty"`
Version string `json:"version,omitempty"`
ServerCert []*x509.Certificate `json:"-"`
Id string `json:"id,omitempty"`
Addr string `json:"addr,omitempty"`
Version string `json:"version,omitempty"`
ServerCert []*x509.Certificate `json:"-"`
ApiAddresses map[string][]ApiAddress `json:"apiAddresses"`
}
type ApiAddress struct {
Url string `json:"url"`
Version string `json:"version"`
}
func (self *ClusterPeer) String() string {
controller/internal/routes/controller_api_model.go
+43 -9
View File
@@ -18,6 +18,7 @@ package routes
import (
"github.com/openziti/edge-api/rest_model"
edgeController "github.com/openziti/ziti/controller"
"github.com/openziti/ziti/controller/env"
"github.com/openziti/ziti/controller/model"
"github.com/openziti/ziti/controller/response"
@@ -27,23 +28,56 @@ const EntityNameController = "controllers"
var ControllerLinkFactory = NewBasicLinkFactory(EntityNameController)
func MapControllerToRestEntity(_ *env.AppEnv, _ *response.RequestContext, Controller *model.Controller) (interface{}, error) {
return MapControllerToRestModel(Controller)
func MapControllerToManagementRestEntity(_ *env.AppEnv, _ *response.RequestContext, Controller *model.Controller) (interface{}, error) {
return MapControllerToManagementRestModel(Controller)
}
func MapControllerToRestModel(controller *model.Controller) (*rest_model.ControllerDetail, error) {
func MapControllerToClientRestEntity(_ *env.AppEnv, _ *response.RequestContext, Controller *model.Controller) (interface{}, error) {
return MapControllerToClientRestModel(Controller)
}
func MapControllerToManagementRestModel(controller *model.Controller) (*rest_model.ControllerDetail, error) {
ret := &rest_model.ControllerDetail{
BaseEntity: BaseEntityToRestModel(controller, ControllerLinkFactory),
Name: &controller.Name,
Address: &controller.Address,
CertPem: &controller.CertPem,
Fingerprint: &controller.Fingerprint,
IsOnline: &controller.IsOnline,
BaseEntity: BaseEntityToRestModel(controller, ControllerLinkFactory),
Name: &controller.Name,
CtrlAddress: &controller.CtrlAddress,
APIAddresses: rest_model.APIAddressList{},
CertPem: &controller.CertPem,
Fingerprint: &controller.Fingerprint,
IsOnline: &controller.IsOnline,
}
if controller.LastJoinedAt != nil {
ret.LastJoinedAt = toStrFmtDateTimeP(*controller.LastJoinedAt)
}
for apiKey, instances := range controller.ApiAddresses {
ret.APIAddresses[apiKey] = rest_model.APIAddressArray{}
for _, instance := range instances {
ret.APIAddresses[apiKey] = append(ret.APIAddresses[apiKey], &rest_model.APIAddress{
URL: instance.Url,
Version: instance.Version,
})
}
}
return ret, nil
}
func MapControllerToClientRestModel(controller *model.Controller) (*rest_model.ControllerDetail, error) {
ret, err := MapControllerToManagementRestModel(controller)
if err != nil {
return nil, err
}
ret.CtrlAddress = nil
for apiKey := range ret.APIAddresses {
if apiKey != edgeController.ClientApiBinding && apiKey != edgeController.OidcApiBinding {
delete(ret.APIAddresses, apiKey)
}
}
return ret, nil
}
controller/internal/routes/controller_router.go
+8 -4
View File
@@ -43,14 +43,18 @@ func NewControllerRouter() *ControllerRouter {
func (r *ControllerRouter) Register(ae *env.AppEnv) {
ae.ManagementApi.ControllersListControllersHandler = controllersMan.ListControllersHandlerFunc(func(params controllersMan.ListControllersParams, _ interface{}) middleware.Responder {
return ae.IsAllowed(r.List, params.HTTPRequest, "", "", permissions.IsAdmin())
return ae.IsAllowed(r.ListManagement, params.HTTPRequest, "", "", permissions.IsAuthenticated())
})
ae.ClientApi.ControllersListControllersHandler = controllersClient.ListControllersHandlerFunc(func(params controllersClient.ListControllersParams, _ interface{}) middleware.Responder {
return ae.IsAllowed(r.List, params.HTTPRequest, "", "", permissions.IsAdmin())
return ae.IsAllowed(r.ListClient, params.HTTPRequest, "", "", permissions.IsAuthenticated())
})
}
func (r *ControllerRouter) List(ae *env.AppEnv, rc *response.RequestContext) {
ListWithHandler[*model.Controller](ae, rc, ae.Managers.Controller, MapControllerToRestEntity)
func (r *ControllerRouter) ListManagement(ae *env.AppEnv, rc *response.RequestContext) {
ListWithHandler[*model.Controller](ae, rc, ae.Managers.Controller, MapControllerToManagementRestEntity)
}
func (r *ControllerRouter) ListClient(ae *env.AppEnv, rc *response.RequestContext) {
ListWithHandler[*model.Controller](ae, rc, ae.Managers.Controller, MapControllerToClientRestEntity)
}
controller/model/controller_manager.go
+105 -12
View File
@@ -17,6 +17,7 @@
package model
import (
"crypto/x509"
"github.com/michaelquigley/pfxlog"
nfpem "github.com/openziti/foundation/v2/pem"
"github.com/openziti/foundation/v2/stringz"
@@ -117,11 +118,22 @@ func (self *ControllerManager) Marshall(entity *Controller) ([]byte, error) {
msg := &edge_cmd_pb.Controller{
Id: entity.Id,
Name: entity.Name,
Address: entity.Address,
Address: entity.CtrlAddress,
CertPem: entity.CertPem,
Fingerprint: entity.Fingerprint,
IsOnline: entity.IsOnline,
LastJoinedAt: timePtrToPb(entity.LastJoinedAt),
ApiAddresses: map[string]*edge_cmd_pb.ApiAddressList{},
}
for apiKey, instances := range entity.ApiAddresses {
msg.ApiAddresses[apiKey] = &edge_cmd_pb.ApiAddressList{}
for _, instance := range instances {
msg.ApiAddresses[apiKey].Addresses = append(msg.ApiAddresses[apiKey].Addresses, &edge_cmd_pb.ApiAddress{
Url: instance.Url,
Version: instance.Version,
})
}
}
return proto.Marshal(msg)
@@ -134,18 +146,57 @@ func (self *ControllerManager) Unmarshall(bytes []byte) (*Controller, error) {
return nil, err
}
return &Controller{
controller := &Controller{
BaseEntity: models.BaseEntity{
Id: msg.Id,
Tags: edge_cmd_pb.DecodeTags(msg.Tags),
},
Name: msg.Name,
Address: msg.Address,
CtrlAddress: msg.Address,
CertPem: msg.CertPem,
Fingerprint: msg.Fingerprint,
IsOnline: msg.IsOnline,
LastJoinedAt: pbTimeToTimePtr(msg.LastJoinedAt),
}, nil
ApiAddresses: map[string][]ApiAddress{},
}
for apiKey, instanceList := range msg.ApiAddresses {
controller.ApiAddresses[apiKey] = nil
if instanceList != nil {
for _, instance := range instanceList.Addresses {
controller.ApiAddresses[apiKey] = append(controller.ApiAddresses[apiKey], ApiAddress{
Url: instance.Url,
Version: instance.Version,
})
}
}
}
return controller, nil
}
func (self *ControllerManager) getCurrentAsClusterPeer() *event.ClusterPeer {
addr, id, version := self.env.GetHostController().GetRaftInfo()
tlsConfig, _, _ := self.env.GetServerCert()
var leaderCerts []*x509.Certificate
for _, certBytes := range tlsConfig.Certificate {
cert, err := x509.ParseCertificate(certBytes)
if err == nil {
leaderCerts = append(leaderCerts, cert)
}
}
apiAddresses, _ := self.env.GetHostController().GetApiAddresses()
return &event.ClusterPeer{
Id: id,
Addr: addr,
Version: version,
ServerCert: leaderCerts,
ApiAddresses: apiAddresses,
}
}
func (self *ControllerManager) PeersConnected(peers []*event.ClusterPeer) {
@@ -165,27 +216,58 @@ func (self *ControllerManager) PeersConnected(peers []*event.ClusterPeer) {
}
connectFields := fields.UpdatedFieldsMap{
db.FieldControllerLastJoinedAt: struct{}{},
db.FieldControllerCertPem: struct{}{},
db.FieldControllerFingerprint: struct{}{},
db.FieldControllerIsOnline: struct{}{},
db.FieldControllerAddress: struct{}{},
db.FieldControllerLastJoinedAt: struct{}{},
db.FieldControllerCertPem: struct{}{},
db.FieldControllerFingerprint: struct{}{},
db.FieldControllerIsOnline: struct{}{},
db.FieldControllerCtrlAddress: struct{}{},
db.FieldControllerApiAddresses: struct{}{},
db.FieldControllerApiAddressUrl: struct{}{},
db.FieldControllerApiAddressVersion: struct{}{},
}
now := time.Now()
selfAsPeer := self.getCurrentAsClusterPeer()
peerFingerprints := ""
for _, peer := range peers {
fingerprint := nfpem.FingerprintFromCertificate(peer.ServerCert[0])
if peerFingerprints == "" {
peerFingerprints = fingerprint
} else {
peerFingerprints = peerFingerprints + ", " + fingerprint
}
}
pfxlog.Logger().Infof("acting as leader, updating controllers with peers, self: %s, peers: %s", nfpem.FingerprintFromCertificate(selfAsPeer.ServerCert[0]), peerFingerprints)
// always as this controller as a "peer" to add or update on the controller list
peers = append(peers, selfAsPeer)
for _, peer := range peers {
if stringz.Contains(controllerIds, peer.Id) {
existing, err := self.Read(peer.Id)
if err != nil {
pfxlog.Logger().WithError(err).Error("could not handle new peer(s) connection, existing controller could not be read")
continue
}
existing.Address = peer.Addr
existing.CtrlAddress = peer.Addr
existing.IsOnline = true
existing.LastJoinedAt = &now
existing.ApiAddresses = map[string][]ApiAddress{}
for apiKey, instances := range peer.ApiAddresses {
existing.ApiAddresses[apiKey] = nil
for _, instance := range instances {
existing.ApiAddresses[apiKey] = append(existing.ApiAddresses[apiKey], ApiAddress{
Url: instance.Url,
Version: instance.Version,
})
}
}
if len(peer.ServerCert) > 0 {
existing.CertPem = nfpem.EncodeToString(peer.ServerCert[0])
@@ -205,11 +287,22 @@ func (self *ControllerManager) PeersConnected(peers []*event.ClusterPeer) {
BaseEntity: models.BaseEntity{
Id: peer.Id,
},
Address: peer.Addr,
CtrlAddress: peer.Addr,
IsOnline: true,
LastJoinedAt: &now,
}
for apiKey, instances := range peer.ApiAddresses {
newController.ApiAddresses[apiKey] = nil
for _, instance := range instances {
newController.ApiAddresses[apiKey] = append(newController.ApiAddresses[apiKey], ApiAddress{
Url: instance.Url,
Version: instance.Version,
})
}
}
newController.Name = peer.ServerCert[0].Subject.CommonName
newController.CertPem = nfpem.EncodeToString(peer.ServerCert[0])
newController.Fingerprint = nfpem.FingerprintFromCertificate(peer.ServerCert[0])
controller/model/controller_model.go
+31 -3
View File
@@ -27,22 +27,39 @@ import (
type Controller struct {
models.BaseEntity
Name string
Address string
CtrlAddress string
CertPem string
Fingerprint string
IsOnline bool
LastJoinedAt *time.Time
ApiAddresses map[string][]ApiAddress
}
type ApiAddress struct {
Url string `json:"url"`
Version string `json:"version"`
}
func (entity *Controller) toBoltEntity(tx *bbolt.Tx, env Env) (*db.Controller, error) {
boltEntity := &db.Controller{
BaseExtEntity: *boltz.NewExtEntity(entity.Id, entity.Tags),
Name: entity.Name,
Address: entity.Address,
CtrlAddress: entity.CtrlAddress,
CertPem: entity.CertPem,
Fingerprint: entity.Fingerprint,
IsOnline: entity.IsOnline,
LastJoinedAt: entity.LastJoinedAt,
ApiAddresses: map[string][]db.ApiAddress{},
}
for apiKey, instances := range entity.ApiAddresses {
boltEntity.ApiAddresses[apiKey] = nil
for _, instance := range instances {
boltEntity.ApiAddresses[apiKey] = append(boltEntity.ApiAddresses[apiKey], db.ApiAddress{
Url: instance.Url,
Version: instance.Version,
})
}
}
return boltEntity, nil
@@ -59,11 +76,22 @@ func (entity *Controller) toBoltEntityForUpdate(tx *bbolt.Tx, env Env, _ boltz.F
func (entity *Controller) fillFrom(env Env, tx *bbolt.Tx, boltController *db.Controller) error {
entity.FillCommon(boltController)
entity.Name = boltController.Name
entity.Address = boltController.Address
entity.CtrlAddress = boltController.CtrlAddress
entity.CertPem = boltController.CertPem
entity.Fingerprint = boltController.Fingerprint
entity.IsOnline = boltController.IsOnline
entity.LastJoinedAt = boltController.LastJoinedAt
entity.ApiAddresses = map[string][]ApiAddress{}
for apiKey, instances := range boltController.ApiAddresses {
entity.ApiAddresses[apiKey] = nil
for _, instance := range instances {
entity.ApiAddresses[apiKey] = append(entity.ApiAddresses[apiKey], ApiAddress{
Url: instance.Url,
Version: instance.Version,
})
}
}
return nil
}
controller/model/env.go
+3
View File
@@ -26,6 +26,7 @@ import (
"github.com/openziti/ziti/common/cert"
"github.com/openziti/ziti/controller/config"
"github.com/openziti/ziti/controller/db"
"github.com/openziti/ziti/controller/event"
"github.com/openziti/ziti/controller/jwtsigner"
"github.com/openziti/ziti/controller/network"
"github.com/xeipuuv/gojsonschema"
@@ -67,6 +68,8 @@ type HostController interface {
Identity() identity.Identity
GetPeerSigners() []*x509.Certificate
GetRaftIndex() uint64
GetRaftInfo() (string, string, string)
GetApiAddresses() (map[string][]event.ApiAddress, []byte)
}
type Schemas interface {
controller/model/testing.go
+9 -1
View File
@@ -50,6 +50,14 @@ type testHostController struct {
ctx *TestContext
}
func (self *testHostController) GetApiAddresses() (map[string][]event.ApiAddress, []byte) {
return nil, nil
}
func (self *testHostController) GetRaftInfo() (string, string, string) {
return "testaddr", "testid", "testversion"
}
func (self *testHostController) GetRaftIndex() uint64 {
return 0
}
@@ -171,7 +179,7 @@ func (ctx *TestContext) GetApiServerCsrSigner() cert.Signer {
}
func (ctx *TestContext) GetControlClientCsrSigner() cert.Signer {
panic("implement me")
return nil
}
func (ctx *TestContext) GetHostController() HostController {
controller/raft/mesh/mesh.go
+52 -14
View File
@@ -18,6 +18,7 @@ package mesh
import (
"crypto/x509"
"encoding/json"
"fmt"
"github.com/openziti/foundation/v2/concurrenz"
"github.com/openziti/foundation/v2/versions"
@@ -41,9 +42,10 @@ import (
const (
PeerAddrHeader = 11
RaftConnectType = 2048
RaftDataType = 2049
SigningCertHeader = 2050
RaftConnectType = 2048
RaftDataType = 2049
SigningCertHeader = 2050
ApiAddressesHeader = 2051
ChannelTypeMesh = "ctrl.mesh"
)
@@ -56,6 +58,7 @@ type Peer struct {
RaftConn *raftPeerConn
Version *versions.VersionInfo
SigningCerts []*x509.Certificate
ApiAddresses map[string][]event.ApiAddress
}
func (self *Peer) HandleClose(channel.Channel) {
@@ -143,7 +146,7 @@ type Mesh interface {
Init(bindHandler channel.BindHandler)
}
func New(env Env, raftAddr raft.ServerAddress) Mesh {
func New(env Env, raftAddr raft.ServerAddress, helloHeaderProviders []HeaderProvider) Mesh {
versionEncoded, err := env.GetVersionProvider().EncoderDecoder().Encode(env.GetVersionProvider().AsVersionInfo())
if err != nil {
panic(err)
@@ -156,12 +159,13 @@ func New(env Env, raftAddr raft.ServerAddress) Mesh {
network: "mesh",
addr: string(raftAddr),
},
Peers: map[string]*Peer{},
closeNotify: make(chan struct{}),
raftAccepts: make(chan net.Conn),
version: env.GetVersionProvider(),
versionEncoded: versionEncoded,
eventDispatcher: env.GetEventDispatcher(),
Peers: map[string]*Peer{},
closeNotify: make(chan struct{}),
raftAccepts: make(chan net.Conn),
version: env.GetVersionProvider(),
versionEncoded: versionEncoded,
eventDispatcher: env.GetEventDispatcher(),
helloHeaderProviders: helloHeaderProviders,
}
}
@@ -180,6 +184,7 @@ type impl struct {
readonly atomic.Bool
clusterStateHandlers concurrenz.CopyOnWriteSlice[func(state ClusterState)]
eventDispatcher event.Dispatcher
helloHeaderProviders []HeaderProvider
}
func (self *impl) RegisterClusterStateHandler(f func(state ClusterState)) {
@@ -257,6 +262,10 @@ func (self *impl) GetOrConnectPeer(address string, timeout time.Duration) (*Peer
SigningCertHeader: serverCert,
}
for _, headerProvider := range self.helloHeaderProviders {
headerProvider.Apply(headers)
}
dialer := channel.NewClassicDialer(self.id, addr, headers)
dialOptions := channel.DefaultOptions()
dialOptions.ConnectOptions.ConnectTimeout = timeout
@@ -293,6 +302,13 @@ func (self *impl) GetOrConnectPeer(address string, timeout time.Duration) (*Peer
return errors.Wrap(err, "can't decode version from returned from peer")
}
if apiAddressBytes, found := peer.Channel.Underlay().Headers()[ApiAddressesHeader]; found {
err := json.Unmarshal(apiAddressBytes, &peer.ApiAddresses)
if err != nil {
pfxlog.Logger().WithError(err).Error("could not unmarshal api address header")
}
}
peer.Version = versionInfo
peer.RaftConn = newRaftPeerConn(peer, self.netAddr)
peer.SigningCerts = []*x509.Certificate{underlay.Certificates()[0]}
@@ -331,6 +347,10 @@ func (self *impl) GetPeerInfo(address string, timeout time.Duration) (raft.Serve
PeerAddrHeader: []byte(self.raftAddr),
}
for _, headerProvider := range self.helloHeaderProviders {
headerProvider.Apply(headers)
}
dialer := channel.NewClassicDialer(self.id, addr, headers)
dialOptions := channel.DefaultOptions()
dialOptions.ConnectOptions.ConnectTimeout = timeout
@@ -404,10 +424,11 @@ func (self *impl) PeerConnected(peer *Peer) error {
evt := event.NewClusterEvent(event.ClusterPeerConnected)
evt.Peers = append(evt.Peers, &event.ClusterPeer{
Id: string(peer.Id),
Addr: peer.Address,
Version: peer.Version.Version,
ServerCert: peer.SigningCerts,
Id: string(peer.Id),
Addr: peer.Address,
Version: peer.Version.Version,
ServerCert: peer.SigningCerts,
ApiAddresses: peer.ApiAddresses,
})
self.eventDispatcher.AcceptClusterEvent(evt)
@@ -522,6 +543,13 @@ func (self *impl) AcceptUnderlay(underlay channel.Underlay) error {
}
}
apiAddressesHeader, found := ch.Underlay().Headers()[ApiAddressesHeader]
if found {
if err := json.Unmarshal(apiAddressesHeader, &peer.ApiAddresses); err != nil {
pfxlog.Logger().WithError(err).Error("could not parse peer api addresses header")
}
}
peer.Version = versionInfo
peer.RaftConn = newRaftPeerConn(peer, self.netAddr)
@@ -560,3 +588,13 @@ func (self *impl) GetPeers() map[string]*Peer {
func (self *impl) IsReadOnly() bool {
return self.readonly.Load()
}
type HeaderProvider interface {
Apply(map[int32][]byte)
}
type HeaderProviderFunc func(map[int32][]byte)
func (self HeaderProviderFunc) Apply(headers map[int32][]byte) {
self(headers)
}
controller/raft/raft.go
+4 -1
View File
@@ -197,6 +197,7 @@ type Env interface {
GetMetricsRegistry() metrics.Registry
GetEventDispatcher() event.Dispatcher
GetCloseNotify() <-chan struct{}
GetHelloHeaderProviders() []mesh.HeaderProvider
}
func NewController(env Env, migrationMgr MigrationManager) *Controller {
@@ -540,7 +541,9 @@ func (self *Controller) Init() error {
return err
}
self.Mesh = mesh.New(self.env, localAddr)
helloHeaderProviders := self.env.GetHelloHeaderProviders()
self.Mesh = mesh.New(self.env, localAddr, helloHeaderProviders)
self.Mesh.RegisterClusterStateHandler(func(state mesh.ClusterState) {
obs := raft.Observation{
Raft: self.Raft,
controller/sync_strats/sync_instant.go
+28 -21
View File
@@ -113,6 +113,7 @@ type InstantStrategy struct {
serviceHandler *constraintToIndexedEvents[*db.EdgeService]
caHandler *constraintToIndexedEvents[*db.Ca]
revocationHandler *constraintToIndexedEvents[*db.Revocation]
controllerHandler *constraintToIndexedEvents[*db.Controller]
}
func (strategy *InstantStrategy) AddPublicKey(cert *tls.Certificate) {
@@ -202,6 +203,12 @@ func (strategy *InstantStrategy) Initialize(logSize uint64, bufferSize uint) err
}
strategy.ae.GetStores().Revocation.AddEntityConstraint(strategy.revocationHandler)
strategy.controllerHandler = &constraintToIndexedEvents[*db.Controller]{
indexProvider: strategy.indexProvider,
createHandler: strategy.ControllerCreate,
updateHandler: strategy.ControllerUpdate,
}
return nil
}
@@ -829,14 +836,22 @@ func (strategy *InstantStrategy) BuildServicePolicies(tx *bbolt.Tx) error {
}
func (strategy *InstantStrategy) BuildPublicKeys(tx *bbolt.Tx) error {
for _, x509Cert := range strategy.ae.HostController.GetPeerSigners() {
publicKey := newPublicKey(x509Cert.Raw, edge_ctrl_pb.DataState_PublicKey_X509CertDer, []edge_ctrl_pb.DataState_PublicKey_Usage{edge_ctrl_pb.DataState_PublicKey_ClientX509CertValidation, edge_ctrl_pb.DataState_PublicKey_JWTValidation})
newModel := &edge_ctrl_pb.DataState_Event_PublicKey{PublicKey: publicKey}
for cursor := strategy.ae.GetStores().Controller.IterateIds(tx, ast.BoolNodeTrue); cursor.IsValid(); cursor.Next() {
currentBytes := cursor.Current()
currentId := string(currentBytes)
storeModel, err := strategy.ae.GetStores().Controller.LoadById(tx, currentId)
if err != nil {
return err
}
certs := nfPem.PemStringToCertificates(storeModel.CertPem)
newModel := &edge_ctrl_pb.DataState_Event_PublicKey{PublicKey: newPublicKey(certs[0].Raw, edge_ctrl_pb.DataState_PublicKey_X509CertDer, []edge_ctrl_pb.DataState_PublicKey_Usage{edge_ctrl_pb.DataState_PublicKey_JWTValidation, edge_ctrl_pb.DataState_PublicKey_ClientX509CertValidation})}
newEvent := &edge_ctrl_pb.DataState_Event{
Action: edge_ctrl_pb.DataState_Create,
Model: newModel,
}
strategy.HandlePublicKeyEvent(newEvent, newModel)
}
@@ -1244,24 +1259,16 @@ func (strategy *InstantStrategy) PostureCheckDelete(index uint64, postureCheck *
strategy.handlePostureCheck(index, edge_ctrl_pb.DataState_Delete, postureCheck)
}
func (strategy *InstantStrategy) PeerAdded(peer *db.Controller) {
if len(peer.CertPem) > 0 {
certs := nfPem.PemBytesToCertificates([]byte(peer.CertPem))
func (strategy *InstantStrategy) ControllerCreate(index uint64, controller *db.Controller) {
certs := nfPem.PemStringToCertificates(controller.CertPem)
cert := certs[0]
strategy.handlePublicKey(index, edge_ctrl_pb.DataState_Create, newPublicKey(cert.Raw, edge_ctrl_pb.DataState_PublicKey_X509CertDer, []edge_ctrl_pb.DataState_PublicKey_Usage{edge_ctrl_pb.DataState_PublicKey_ClientX509CertValidation, edge_ctrl_pb.DataState_PublicKey_JWTValidation}))
}
if len(certs) > 0 {
strategy.Apply(&edge_ctrl_pb.DataState_Event{
IsSynthetic: true,
Action: edge_ctrl_pb.DataState_Create,
Model: &edge_ctrl_pb.DataState_Event_PublicKey{
PublicKey: newPublicKey(certs[0].Raw, edge_ctrl_pb.DataState_PublicKey_X509CertDer, []edge_ctrl_pb.DataState_PublicKey_Usage{edge_ctrl_pb.DataState_PublicKey_ClientX509CertValidation, edge_ctrl_pb.DataState_PublicKey_JWTValidation}),
},
})
} else {
pfxlog.Logger().Warn("peer added event, certificate pem did not parse to at least one certificate")
}
} else {
pfxlog.Logger().Warn("peer added event without certificate pem")
}
func (strategy *InstantStrategy) ControllerUpdate(index uint64, controller *db.Controller) {
certs := nfPem.PemStringToCertificates(controller.CertPem)
cert := certs[0]
strategy.handlePublicKey(index, edge_ctrl_pb.DataState_Create, newPublicKey(cert.Raw, edge_ctrl_pb.DataState_PublicKey_X509CertDer, []edge_ctrl_pb.DataState_PublicKey_Usage{edge_ctrl_pb.DataState_PublicKey_ClientX509CertValidation, edge_ctrl_pb.DataState_PublicKey_JWTValidation}))
}
func (strategy *InstantStrategy) CaCreate(index uint64, ca *db.Ca) {
doc/ha/.gitignore
+5
View File
@@ -0,0 +1,5 @@
er1.*
er1/
er2.*
er2/
endpoints
doc/ha/ctrl1.yml
+5
View File
@@ -51,3 +51,8 @@ web:
- binding: edge-management
- binding: edge-client
- binding: edge-oidc
options:
redirectURIs:
- "http://localhost:*/auth/callback"
- "http://127.0.0.1:*/auth/callback"
- "https://oauth.pstmn.io/v1/callback"
doc/ha/ctrl2.yml
+5
View File
@@ -40,3 +40,8 @@ web:
- binding: edge-management
- binding: edge-client
- binding: edge-oidc
options:
redirectURIs:
- "http://localhost:*/auth/callback"
- "http://127.0.0.1:*/auth/callback"
- "https://oauth.pstmn.io/v1/callback"
doc/ha/ctrl3.yml
+5
View File
@@ -40,3 +40,8 @@ web:
- binding: edge-management
- binding: edge-client
- binding: edge-oidc
options:
redirectURIs:
- "http://localhost:*/auth/callback"
- "http://127.0.0.1:*/auth/callback"
- "https://oauth.pstmn.io/v1/callback"
go.mod
+1 -1
View File
@@ -49,7 +49,7 @@ require (
github.com/natefinch/lumberjack v2.0.0+incompatible
github.com/openziti/agent v1.0.16
github.com/openziti/channel/v2 v2.0.122
github.com/openziti/edge-api v0.26.13
github.com/openziti/edge-api v0.26.14
github.com/openziti/foundation/v2 v2.0.40
github.com/openziti/identity v1.0.73
github.com/openziti/jwks v1.0.3
go.sum
+2 -2
View File
@@ -570,8 +570,8 @@ github.com/openziti/channel/v2 v2.0.122 h1:MPALuLUuTsW2QJzaoRkrP/wGqhIeWtSsrnxt2
github.com/openziti/channel/v2 v2.0.122/go.mod h1:n4r/pWTFYJy98hEtf4ATYVHU70OePjoM8JJzrIvXvZg=
github.com/openziti/dilithium v0.3.3 h1:PLgQ6PMNLSTzCFbX/h98cmudgz/cU6TmjdSv5NAPD8k=
github.com/openziti/dilithium v0.3.3/go.mod h1:vsCjI2AU/hon9e+dLhUFbCNGesJDj2ASgkySOcpmvjo=
github.com/openziti/edge-api v0.26.13 h1:lWAHu04nk7a0AEnJBjmwijUIEE+NvNAiOcdg11dxbIY=
github.com/openziti/edge-api v0.26.13/go.mod h1:aTRqB9J/6ISbw9+YWjzD0tvWAXY98m8yWcbiGNqAQTU=
github.com/openziti/edge-api v0.26.14 h1:BpxKavDVdykOY7GSdKXi7klv9Im4vw7bTMujz3wFE1Q=
github.com/openziti/edge-api v0.26.14/go.mod h1:aTRqB9J/6ISbw9+YWjzD0tvWAXY98m8yWcbiGNqAQTU=
github.com/openziti/foundation/v2 v2.0.40 h1:8oXL77egZjnQuKuxPTHiBOh+X/I2mSJefIbnLSv8rFI=
github.com/openziti/foundation/v2 v2.0.40/go.mod h1:38RikdtjvzDUALm3jT3PSOrU0bHCleRty6bHGdfFyUI=
github.com/openziti/identity v1.0.73 h1:F1iF/rDQYfDIv4ktqBYwK/0C5Ra/7y92Z5Umrw4c4fk=
zititest/go.mod
+1 -1
View File
@@ -14,7 +14,7 @@ require (
github.com/michaelquigley/pfxlog v0.6.10
github.com/openziti/agent v1.0.16
github.com/openziti/channel/v2 v2.0.122
github.com/openziti/edge-api v0.26.13
github.com/openziti/edge-api v0.26.14
github.com/openziti/fablab v0.5.42
github.com/openziti/foundation/v2 v2.0.40
github.com/openziti/identity v1.0.73
zititest/go.sum
+2 -2
View File
@@ -592,8 +592,8 @@ github.com/openziti/channel/v2 v2.0.122 h1:MPALuLUuTsW2QJzaoRkrP/wGqhIeWtSsrnxt2
github.com/openziti/channel/v2 v2.0.122/go.mod h1:n4r/pWTFYJy98hEtf4ATYVHU70OePjoM8JJzrIvXvZg=
github.com/openziti/dilithium v0.3.3 h1:PLgQ6PMNLSTzCFbX/h98cmudgz/cU6TmjdSv5NAPD8k=
github.com/openziti/dilithium v0.3.3/go.mod h1:vsCjI2AU/hon9e+dLhUFbCNGesJDj2ASgkySOcpmvjo=
github.com/openziti/edge-api v0.26.13 h1:lWAHu04nk7a0AEnJBjmwijUIEE+NvNAiOcdg11dxbIY=
github.com/openziti/edge-api v0.26.13/go.mod h1:aTRqB9J/6ISbw9+YWjzD0tvWAXY98m8yWcbiGNqAQTU=
github.com/openziti/edge-api v0.26.14 h1:BpxKavDVdykOY7GSdKXi7klv9Im4vw7bTMujz3wFE1Q=
github.com/openziti/edge-api v0.26.14/go.mod h1:aTRqB9J/6ISbw9+YWjzD0tvWAXY98m8yWcbiGNqAQTU=
github.com/openziti/fablab v0.5.42 h1:vENJKfEba2T4sSLwlKDL/IzBYfY8iHnhc4umf6IESiY=
github.com/openziti/fablab v0.5.42/go.mod h1:HDT06y1QX8kO8ZQrgHvZmJsvc8iRybESGtlDLDII4ks=
github.com/openziti/foundation/v2 v2.0.40 h1:8oXL77egZjnQuKuxPTHiBOh+X/I2mSJefIbnLSv8rFI=