mirror of
https://github.com/zitadel/zitadel.git
synced 2025-08-12 10:57:35 +00:00
fix(queries): authn keys (#2820)
* begin authn keys * single table for state change * add key type * begin authn keys query * query * tests * fix merge * remove wrong migration version * improve filter * Update projection.go * cleanup
This commit is contained in:
@@ -1,124 +0,0 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"github.com/caos/zitadel/internal/eventstore/v1"
|
||||
"time"
|
||||
|
||||
"github.com/caos/logging"
|
||||
|
||||
es_models "github.com/caos/zitadel/internal/eventstore/v1/models"
|
||||
"github.com/caos/zitadel/internal/eventstore/v1/query"
|
||||
"github.com/caos/zitadel/internal/eventstore/v1/spooler"
|
||||
key_model "github.com/caos/zitadel/internal/key/repository/view/model"
|
||||
proj_model "github.com/caos/zitadel/internal/project/repository/eventsourcing/model"
|
||||
user_model "github.com/caos/zitadel/internal/user/repository/eventsourcing/model"
|
||||
)
|
||||
|
||||
const (
|
||||
authnKeysTable = "auth.authn_keys"
|
||||
)
|
||||
|
||||
type AuthNKeys struct {
|
||||
handler
|
||||
subscription *v1.Subscription
|
||||
}
|
||||
|
||||
func newAuthNKeys(handler handler) *AuthNKeys {
|
||||
h := &AuthNKeys{
|
||||
handler: handler,
|
||||
}
|
||||
|
||||
h.subscribe()
|
||||
|
||||
return h
|
||||
}
|
||||
|
||||
func (k *AuthNKeys) subscribe() {
|
||||
k.subscription = k.es.Subscribe(k.AggregateTypes()...)
|
||||
go func() {
|
||||
for event := range k.subscription.Events {
|
||||
query.ReduceEvent(k, event)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (k *AuthNKeys) ViewModel() string {
|
||||
return authnKeysTable
|
||||
}
|
||||
|
||||
func (k *AuthNKeys) Subscription() *v1.Subscription {
|
||||
return k.subscription
|
||||
}
|
||||
|
||||
func (_ *AuthNKeys) AggregateTypes() []es_models.AggregateType {
|
||||
return []es_models.AggregateType{user_model.UserAggregate, proj_model.ProjectAggregate}
|
||||
}
|
||||
|
||||
func (k *AuthNKeys) CurrentSequence() (uint64, error) {
|
||||
sequence, err := k.view.GetLatestAuthNKeySequence()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return sequence.CurrentSequence, nil
|
||||
}
|
||||
|
||||
func (k *AuthNKeys) EventQuery() (*es_models.SearchQuery, error) {
|
||||
sequence, err := k.view.GetLatestAuthNKeySequence()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return es_models.NewSearchQuery().
|
||||
AggregateTypeFilter(k.AggregateTypes()...).
|
||||
LatestSequenceFilter(sequence.CurrentSequence), nil
|
||||
}
|
||||
|
||||
func (k *AuthNKeys) Reduce(event *es_models.Event) (err error) {
|
||||
switch event.AggregateType {
|
||||
case user_model.UserAggregate,
|
||||
proj_model.ProjectAggregate:
|
||||
err = k.processAuthNKeys(event)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (k *AuthNKeys) processAuthNKeys(event *es_models.Event) (err error) {
|
||||
key := new(key_model.AuthNKeyView)
|
||||
switch event.Type {
|
||||
case user_model.MachineKeyAdded,
|
||||
proj_model.ClientKeyAdded:
|
||||
err = key.AppendEvent(event)
|
||||
if key.ExpirationDate.Before(time.Now()) {
|
||||
return k.view.ProcessedAuthNKeySequence(event)
|
||||
}
|
||||
case user_model.MachineKeyRemoved:
|
||||
err = key.SetUserData(event)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return k.view.DeleteAuthNKey(key.ID, event)
|
||||
case proj_model.ClientKeyRemoved:
|
||||
err = key.SetClientData(event)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return k.view.DeleteAuthNKey(key.ID, event)
|
||||
case user_model.UserRemoved,
|
||||
proj_model.ApplicationRemoved:
|
||||
return k.view.DeleteAuthNKeysByObjectID(event.AggregateID, event)
|
||||
default:
|
||||
return k.view.ProcessedAuthNKeySequence(event)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return k.view.PutAuthNKey(key, event)
|
||||
}
|
||||
|
||||
func (k *AuthNKeys) OnError(event *es_models.Event, err error) error {
|
||||
logging.LogWithFields("SPOOL-S9fe", "id", event.AggregateID).WithError(err).Warn("something went wrong in authn key handler")
|
||||
return spooler.HandleError(event, err, k.view.GetLatestAuthNKeyFailedEvent, k.view.ProcessedAuthNKeyFailedEvent, k.view.ProcessedAuthNKeySequence, k.errorCountUntilSkip)
|
||||
}
|
||||
|
||||
func (k *AuthNKeys) OnSuccess() error {
|
||||
return spooler.HandleSuccess(k.view.UpdateAuthNKeySpoolerRunTimestamp)
|
||||
}
|
@@ -47,8 +47,6 @@ func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, es
|
||||
newUserGrant(
|
||||
handler{view, bulkLimit, configs.cycleDuration("UserGrant"), errorCount, es},
|
||||
systemDefaults.IamID),
|
||||
newAuthNKeys(
|
||||
handler{view, bulkLimit, configs.cycleDuration("MachineKey"), errorCount, es}),
|
||||
newIDPConfig(
|
||||
handler{view, bulkLimit, configs.cycleDuration("IDPConfig"), errorCount, es}),
|
||||
newIDPProvider(
|
||||
|
Reference in New Issue
Block a user