lahirujayathilake commented on code in PR #518: URL: https://github.com/apache/airavata-custos/pull/518#discussion_r3785767288
########## connectors/LDAP/Provisioner/pkg/ldap/loader.go: ########## @@ -0,0 +1,277 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package ldap is the LDAP Provisioner entry point. Wired from +// internal/connectors/loader.go alongside the COmanage connector — a site +// runs one or the other depending on whether it uses COmanage as a +// managed layer in front of the directory. +package ldap + +import ( + "context" + "log/slog" + "os" + "strconv" + "sync" + "time" + + "github.com/jmoiron/sqlx" + + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/client" + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/store" + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/subscribers" + ldapdb "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/db" + "github.com/apache/airavata-custos/internal/config" + "github.com/apache/airavata-custos/internal/db" + "github.com/apache/airavata-custos/internal/tracing" + "github.com/apache/airavata-custos/pkg/events" + "github.com/apache/airavata-custos/pkg/identity" + "github.com/apache/airavata-custos/pkg/service" +) + +const connectorName = "ldap" + +// init registers the event types that close out an audit trace for this +// connector, so the audit-trace UI marks a provisioning run as complete +// instead of leaving it at in_progress. Mirrors the equivalent block in +// the COmanage and AMIE loaders. +func init() { + tracing.RegisterTerminalMarkers("ldap", + "LDAPAccountCreated", + "LDAPAccountUpdated", + "LDAPGroupCreated", + ) +} + +// LoadConnector wires the subscriber to the event bus. Reads YAML config +// first and falls back to environment variables. If neither yields a +// complete config, it logs and returns nil without registering — same +// skip-with-log pattern the other connectors use so a dev server boots +// without LDAP credentials. +func LoadConnector(ctx context.Context, database *sqlx.DB, eventBus *events.Bus, coreService *service.Service, _ *sync.WaitGroup, _ *identity.Router, connectorConfig *config.ConnectorConfig) error { + cfg, ok := loadConfigFromConnectorConfig(connectorConfig) + if !ok { + cfg, ok = loadConfigFromEnv() + if !ok { + slog.Info("ldap provisioner: required config not set; skipping") + return nil + } + } + + if err := db.MigrateConnectorFS(database, ldapdb.MigrationFS(), "migrations", connectorName); err != nil { + return err + } + + ldapClient, err := client.New(cfg) + if err != nil { + return err + } + + uidSeq := store.NewUIDSequence(database) + if err := seedUIDCounter(ctx, ldapClient, uidSeq, cfg); err != nil { + return err + } + + subscribers.NewClusterUserSubscriber(ldapClient, uidSeq, eventBus, coreService, cfg.CustosClusterID).RegisterSubscribers() + slog.Info("ldap provisioner: subscriber registered", + "url", cfg.URL, "base_dn", cfg.BaseDN, "cluster_id", cfg.CustosClusterID) + // Custos is the source of truth for provisioning: entries created by this + // connector must not be edited directly in LDAP. There is no drift + // reconciliation; out-of-band edits will be invisible to Custos. + return nil +} + +// seedUIDCounter initialises the ldap_uid_sequence row for this +// cluster. Runs one LDAP scan to find max(existing uidNumber) so the +// counter starts above any entries provisioned out-of-band before +// this connector ran. Idempotent: on subsequent boots the store's +// GREATEST() upsert preserves whatever value the counter has grown +// to, so a re-scan cannot regress the sequence. +func seedUIDCounter(ctx context.Context, ldapClient *client.Client, seq *store.UIDSequence, cfg client.Config) error { + minUID := cfg.MinUID + if minUID <= 0 { + minUID = client.DefaultMinUID + } + + // One-time LDAP scan to find the current head of the uidNumber + // range. AllocateNextUID(minUID) returns max(uidNumber)+1, floored + // at minUID — exactly the value we want as the counter's next_uid. + // If the scan fails (e.g. LDAP unreachable at boot), fall back to + // the floor so the connector still starts. + initial, err := ldapClient.AllocateNextUID(minUID) + if err != nil { + slog.Warn("ldap provisioner: could not scan LDAP for existing max uidNumber during seed; falling back to floor", + "floor", minUID, "err", err) + initial = minUID + } + + if err := seq.Seed(ctx, cfg.CustosClusterID, initial); err != nil { + return err + } + slog.Info("ldap provisioner: uid sequence seeded", + "cluster_id", cfg.CustosClusterID, "initial_next_uid", initial) + return nil +} + +func loadConfigFromConnectorConfig(connectorConfig *config.ConnectorConfig) (client.Config, bool) { + if connectorConfig == nil { + return client.Config{}, false + } + + var url, bindDN, bindPassword, baseDN, custosCluster, defaultShell, homedirPrefix string + verifySSL := true + timeout := 30 * time.Second + minUID := client.DefaultMinUID + + var groupBaseDN string + + if directory, err := connectorConfig.GetNestedConfig("directory"); err == nil { + if v, ok := directory["url"].(string); ok { + url = v + } + if v, ok := directory["bind_dn"].(string); ok { + bindDN = v + } + if v, ok := directory["bind_password"].(string); ok { + bindPassword = v + } + if v, ok := directory["base_dn"].(string); ok { + baseDN = v + } + if v, ok := directory["group_base_dn"].(string); ok { + groupBaseDN = v + } + if v, ok := directory["verify_ssl"].(bool); ok { + verifySSL = v + } + } + + if provisioning, err := connectorConfig.GetNestedConfig("provisioning"); err == nil { + if v, ok := provisioning["custos_cluster_id"].(string); ok { + custosCluster = v + } + if v, ok := provisioning["default_shell"].(string); ok { + defaultShell = v + } + if v, ok := provisioning["homedir_prefix"].(string); ok { + homedirPrefix = v + } + if v, ok := provisioning["http_timeout"].(string); ok { + if d, err := time.ParseDuration(v); err == nil { + timeout = d + } + } + if n := asInt64(provisioning["min_uid"]); n > 0 { + minUID = n + } + } + + if url == "" || bindDN == "" || bindPassword == "" || baseDN == "" || custosCluster == "" { + return client.Config{}, false + } + + if defaultShell == "" { + defaultShell = "/bin/bash" + } + if homedirPrefix == "" { + homedirPrefix = "/home/" + } + + return client.Config{ + URL: url, + BindDN: bindDN, + BindPassword: bindPassword, + BaseDN: baseDN, + GroupBaseDN: groupBaseDN, + VerifySSL: verifySSL, + CustosClusterID: custosCluster, + DefaultShell: defaultShell, + HomedirPrefix: homedirPrefix, + Timeout: timeout, + MinUID: minUID, + }, true +} + +// asInt64 tolerates the int / int64 / float64 shapes yaml.v3 may produce. +func asInt64(v interface{}) int64 { + switch n := v.(type) { + case int: + return int64(n) + case int64: + return n + case float64: + return int64(n) + } + return 0 +} + +func loadConfigFromEnv() (client.Config, bool) { Review Comment: Drop this and read from the config yaml? Env values get injected into the yaml anyway. ########## connectors/LDAP/Provisioner/internal/client/client.go: ########## @@ -0,0 +1,472 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package client is the LDAP protocol wrapper the LDAP Provisioner uses to +// read and write directory entries. It parallels the REST client the +// COmanage Identity-Provisioner uses; different wire protocol, same +// architectural role. +package client + +import ( + "crypto/tls" + "errors" + "fmt" + "net" + "strconv" + "sync" + "time" + + "github.com/go-ldap/ldap/v3" +) + +// ErrNotFound is returned when a search comes back empty. Mirrors the +// COmanage client's ErrNotFound so orchestration code can use errors.Is. +var ErrNotFound = errors.New("ldap: not found") + +// Config carries the connection parameters and the per-cluster identity +// the connector serves. CustosClusterID lets a subscriber filter events so +// a single deployment can host multiple provisioner instances side by side. +type Config struct { + URL string + BindDN string + BindPassword string + BaseDN string + VerifySSL bool + CustosClusterID string + DefaultShell string + HomedirPrefix string + Timeout time.Duration + + // MinUID is the lowest POSIX uidNumber the allocator will hand out. + // Sites usually reserve 0..999 for system, 1000..49999 for local + // users, and 50000+ for federated / auto-provisioned users. Default + // 50000 when unset. + MinUID int64 + + // GroupBaseDN is the container for posixGroup entries — e.g. + // "ou=groups,dc=example,dc=edu". When empty, the connector skips + // posixGroup creation entirely (fine on systems using automatic + // private groups; needed for strict SSSD setups). + GroupBaseDN string +} + +// DefaultMinUID is the value used when Config.MinUID is zero. Chosen to +// sit above the typical local-user range on RHEL / Debian derivatives. +const DefaultMinUID int64 = 50000 + +// Connection abstracts the go-ldap operations the client uses. Defined here +// so unit tests can substitute a fake without dialing a real server. +type Connection interface { + Bind(username, password string) error + Add(req *ldap.AddRequest) error + Modify(req *ldap.ModifyRequest) error + Search(req *ldap.SearchRequest) (*ldap.SearchResult, error) + Close() error +} + +// Dialer opens a Connection. Injected so tests can bypass real networking. +type Dialer interface { + Dial(url string, verifySSL bool, timeout time.Duration) (Connection, error) +} + +type defaultDialer struct{} + +func (defaultDialer) Dial(url string, verifySSL bool, timeout time.Duration) (Connection, error) { + opts := []ldap.DialOpt{ + ldap.DialWithTLSConfig(&tls.Config{InsecureSkipVerify: !verifySSL}), + } + if timeout > 0 { + opts = append(opts, ldap.DialWithDialer(&net.Dialer{Timeout: timeout})) + } + return ldap.DialURL(url, opts...) +} + +// PosixAccount is the subset of LDAP attributes the connector maintains +// for each user. UIDNumber and GIDNumber are required — schema-conformant +// posixAccount entries must have both. +type PosixAccount struct { + UID string + UIDNumber int64 + GIDNumber int64 + GivenName string + Surname string + Mail string + HomeDirectory string + LoginShell string +} + +// Client is a thin, connection-reusing wrapper over go-ldap. All public +// methods take Client.mu; helpers with the `Locked` suffix assume the +// caller already holds it. +type Client struct { + cfg Config + dialer Dialer + + mu sync.Mutex + conn Connection +} + +// New constructs a Client backed by the real go-ldap dialer. +func New(cfg Config) (*Client, error) { + if cfg.URL == "" || cfg.BindDN == "" || cfg.BaseDN == "" { + return nil, errors.New("client.New: Config requires URL, BindDN, and BaseDN") + } + return &Client{cfg: cfg, dialer: defaultDialer{}}, nil +} + +// NewWithDialer is used by tests to inject a fake dialer. +func NewWithDialer(cfg Config, d Dialer) *Client { + return &Client{cfg: cfg, dialer: d} +} + +// Config returns the config the client was built with. +func (c *Client) Config() Config { return c.cfg } + +// Close releases the underlying connection. +func (c *Client) Close() { + c.mu.Lock() + defer c.mu.Unlock() + c.closeConn() +} + +// FindPosixAccount searches for a posixAccount entry by UID and returns +// its DN plus a map of the retrieved attributes. Returns (empty, nil, nil) +// when no entry matches. +func (c *Client) FindPosixAccount(uid string) (string, map[string][]string, error) { + c.mu.Lock() + defer c.mu.Unlock() + return c.findPosixAccountLocked(uid) +} + +// AddPosixAccount creates a new posixAccount + inetOrgPerson entry. +// Callers supply the uidNumber (typically from internal/store.UIDSequence, +// which is the persistent monotonic allocator that guarantees no reuse +// after entry deletion and serialises cross-process races via InnoDB +// row locking). +func (c *Client) AddPosixAccount(a PosixAccount) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + return c.addPosixAccountLocked(a) +} + +// ModifyPosixAccount replaces the mutable attributes of an existing entry. +// The uid RDN and the numeric IDs are not modified — a change in either +// would be a different account. +func (c *Client) ModifyPosixAccount(dn string, a PosixAccount) error { + c.mu.Lock() + defer c.mu.Unlock() + return c.modifyPosixAccountLocked(dn, a) +} + +// AllocateNextUID scans BaseDN for all posixAccount entries, reads their +// uidNumber attribute, and returns max(uidNumber) + 1, floored at minUID. +// +// Used exclusively for one-time seeding of the persistent uid counter +// at connector startup — the loader calls this so a fresh deployment +// initialises above any entries already in LDAP from out-of-band +// provisioning. Steady-state allocations go through internal/store's +// UIDSequence, which is monotonic across restarts and never regresses +// when LDAP entries are deleted. +func (c *Client) AllocateNextUID(minUID int64) (int64, error) { + c.mu.Lock() + defer c.mu.Unlock() + return c.allocateNextUIDLocked(minUID) +} + +// FindPosixGroup searches for a posixGroup entry by cn under GroupBaseDN +// and returns its DN. Returns "" when no entry matches. Errors when +// GroupBaseDN is empty — callers should check the config first. +func (c *Client) FindPosixGroup(cn string) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.cfg.GroupBaseDN == "" { + return "", errors.New("FindPosixGroup: GroupBaseDN not configured") + } + + conn, err := c.ensureConn() + if err != nil { + return "", err + } + filter := fmt.Sprintf("(&(objectClass=posixGroup)(cn=%s))", ldap.EscapeFilter(cn)) + req := ldap.NewSearchRequest( + c.cfg.GroupBaseDN, + ldap.ScopeWholeSubtree, ldap.NeverDerefAliases, + 2, c.searchTimeoutSeconds(), false, + filter, + []string{"cn", "gidNumber"}, + nil, + ) + result, err := conn.Search(req) + if err != nil { + c.closeConn() + return "", fmt.Errorf("search posixGroup cn=%s: %w", cn, err) + } + if len(result.Entries) == 0 { + return "", nil + } + if len(result.Entries) > 1 { + return "", fmt.Errorf("search posixGroup cn=%s returned %d entries", cn, len(result.Entries)) + } + return result.Entries[0].DN, nil +} + +// AddPosixGroup creates a posixGroup entry at cn=<cn>,<GroupBaseDN>. +// Errors when GroupBaseDN is empty. gidNumber must be positive. +func (c *Client) AddPosixGroup(cn string, gidNumber int64) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.cfg.GroupBaseDN == "" { + return "", errors.New("AddPosixGroup: GroupBaseDN not configured") + } + if cn == "" { + return "", errors.New("AddPosixGroup: cn is required") + } + if gidNumber <= 0 { + return "", errors.New("AddPosixGroup: gidNumber must be a positive integer") + } + + conn, err := c.ensureConn() + if err != nil { + return "", err + } + + dn := fmt.Sprintf("cn=%s,%s", cn, c.cfg.GroupBaseDN) + req := ldap.NewAddRequest(dn, nil) + req.Attribute("objectClass", []string{"top", "posixGroup"}) + req.Attribute("cn", []string{cn}) + req.Attribute("gidNumber", []string{strconv.FormatInt(gidNumber, 10)}) + + if err := conn.Add(req); err != nil { + c.closeConn() + return "", fmt.Errorf("add posixGroup %s: %w", dn, err) + } + return dn, nil +} + +// IsConstraintViolation reports whether err is an LDAP constraint / +// value-already-exists error — the signal that a concurrent writer +// claimed the uidNumber we chose. +func IsConstraintViolation(err error) bool { + return ldap.IsErrorWithCode(err, ldap.LDAPResultConstraintViolation) || + ldap.IsErrorWithCode(err, ldap.LDAPResultAttributeOrValueExists) +} + +// IsAlreadyExists reports whether err is an "entry already exists" LDAP +// error — used by group creation to treat concurrent adds as idempotent. +func IsAlreadyExists(err error) bool { + return ldap.IsErrorWithCode(err, ldap.LDAPResultEntryAlreadyExists) +} + +// ---- private locked helpers ------------------------------------------ + +func (c *Client) findPosixAccountLocked(uid string) (string, map[string][]string, error) { + conn, err := c.ensureConn() + if err != nil { + return "", nil, err + } + filter := fmt.Sprintf("(&(objectClass=posixAccount)(uid=%s))", ldap.EscapeFilter(uid)) + req := ldap.NewSearchRequest( + c.cfg.BaseDN, + ldap.ScopeWholeSubtree, ldap.NeverDerefAliases, + 2, c.searchTimeoutSeconds(), false, + filter, + []string{"uid", "uidNumber", "gidNumber", "cn", "givenName", "sn", "mail", "homeDirectory", "loginShell"}, + nil, + ) + result, err := conn.Search(req) + if err != nil { + c.closeConn() + return "", nil, fmt.Errorf("search posixAccount uid=%s: %w", uid, err) + } + if len(result.Entries) == 0 { + return "", nil, nil + } + if len(result.Entries) > 1 { + return "", nil, fmt.Errorf("search posixAccount uid=%s returned %d entries", uid, len(result.Entries)) + } + entry := result.Entries[0] + attrs := make(map[string][]string, len(entry.Attributes)) + for _, a := range entry.Attributes { + attrs[a.Name] = a.Values + } + return entry.DN, attrs, nil +} + +func (c *Client) addPosixAccountLocked(a PosixAccount) (string, error) { + if err := validate(a); err != nil { + return "", err + } + conn, err := c.ensureConn() + if err != nil { + return "", err + } + + dn := fmt.Sprintf("uid=%s,%s", a.UID, c.cfg.BaseDN) + req := ldap.NewAddRequest(dn, nil) + req.Attribute("objectClass", []string{ + "top", "person", "organizationalPerson", "inetOrgPerson", "posixAccount", + }) + req.Attribute("uid", []string{a.UID}) + req.Attribute("cn", []string{fullName(a)}) + req.Attribute("sn", []string{a.Surname}) + if a.GivenName != "" { + req.Attribute("givenName", []string{a.GivenName}) + } + req.Attribute("uidNumber", []string{strconv.FormatInt(a.UIDNumber, 10)}) + req.Attribute("gidNumber", []string{strconv.FormatInt(a.GIDNumber, 10)}) + req.Attribute("homeDirectory", []string{a.HomeDirectory}) + if a.LoginShell != "" { + req.Attribute("loginShell", []string{a.LoginShell}) + } + if a.Mail != "" { + req.Attribute("mail", []string{a.Mail}) + } + + if err := conn.Add(req); err != nil { + c.closeConn() + return "", fmt.Errorf("add posixAccount %s: %w", dn, err) + } + return dn, nil +} + +func (c *Client) modifyPosixAccountLocked(dn string, a PosixAccount) error { + conn, err := c.ensureConn() + if err != nil { + return err + } + req := ldap.NewModifyRequest(dn, nil) + req.Replace("cn", []string{fullName(a)}) + req.Replace("sn", []string{a.Surname}) + if a.GivenName != "" { + req.Replace("givenName", []string{a.GivenName}) + } + req.Replace("homeDirectory", []string{a.HomeDirectory}) + if a.LoginShell != "" { + req.Replace("loginShell", []string{a.LoginShell}) + } + if a.Mail != "" { + req.Replace("mail", []string{a.Mail}) + } + if err := conn.Modify(req); err != nil { + c.closeConn() + return fmt.Errorf("modify posixAccount %s: %w", dn, err) + } + return nil +} + +func (c *Client) allocateNextUIDLocked(minUID int64) (int64, error) { + if minUID <= 0 { + minUID = DefaultMinUID + } + conn, err := c.ensureConn() + if err != nil { + return 0, err + } + req := ldap.NewSearchRequest( + c.cfg.BaseDN, + ldap.ScopeWholeSubtree, ldap.NeverDerefAliases, + 0, c.searchTimeoutSeconds(), false, + "(&(objectClass=posixAccount)(uidNumber=*))", + []string{"uidNumber"}, + nil, + ) + result, err := conn.Search(req) + if err != nil { + c.closeConn() + return 0, fmt.Errorf("search posixAccount for max uidNumber: %w", err) + } + + var max int64 + for _, entry := range result.Entries { + for _, v := range entry.GetAttributeValues("uidNumber") { + n, err := strconv.ParseInt(v, 10, 64) + if err != nil { + continue + } + if n > max { + max = n + } + } + } + next := max + 1 + if next < minUID { + next = minUID + } + return next, nil +} + +// searchTimeoutSeconds returns Config.Timeout as whole seconds for use +// as the LDAP SearchRequest TimeLimit field. Clamped to a minimum of 1 +// so a sub-second configured timeout does not become 0 (which the LDAP +// protocol interprets as "no time limit" — the opposite of the caller's +// intent). +func (c *Client) searchTimeoutSeconds() int { + n := int(c.cfg.Timeout.Seconds()) + if n < 1 { + return 1 + } + return n +} + +func (c *Client) ensureConn() (Connection, error) { + if c.conn != nil { + return c.conn, nil + } + conn, err := c.dialer.Dial(c.cfg.URL, c.cfg.VerifySSL, c.cfg.Timeout) + if err != nil { + return nil, fmt.Errorf("dial LDAP %s: %w", c.cfg.URL, err) + } + if err := conn.Bind(c.cfg.BindDN, c.cfg.BindPassword); err != nil { + _ = conn.Close() + return nil, fmt.Errorf("bind LDAP as %s: %w", c.cfg.BindDN, err) + } + c.conn = conn + return conn, nil +} + +func (c *Client) closeConn() { + if c.conn != nil { + _ = c.conn.Close() + c.conn = nil + } +} + +func validate(a PosixAccount) error { + if a.UID == "" { + return errors.New("PosixAccount: UID is required") + } + if a.UIDNumber <= 0 { + return errors.New("PosixAccount: UIDNumber must be a positive integer") + } + if a.GIDNumber <= 0 { + return errors.New("PosixAccount: GIDNumber must be a positive integer") + } + if a.HomeDirectory == "" { + return errors.New("PosixAccount: HomeDirectory is required") + } + if a.Surname == "" { Review Comment: Surname is required here and LastName is passed straight through, so a user with no last name fails provisioning. Is that reachable? ########## connectors/LDAP/Provisioner/db/migrations/000001_uid_sequence.up.sql: ########## @@ -0,0 +1,30 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- Persistent monotonic UID allocator per cluster. The row for a given +-- cluster_id holds the NEXT uidNumber to hand out; allocation is a +-- single UPDATE that InnoDB serialises via row locking, so concurrent +-- allocators (in-process or cross-process) cannot pick the same value. +-- +-- Rows are never deleted — decommissioning a cluster is handled at a +-- higher level. Never-decrementing counter is what guarantees no UID +-- reuse after an LDAP entry is deleted. +CREATE TABLE ldap_uid_sequence ( Review Comment: Let's keep these in our own table (connector side) instead of the counter. Then we don't have to cache the `uid` in `user_identities`, or look up the entry by `username` to find it again. COmanage already holds these attributes itself, so this is the same idea on our side. ########## connectors/LDAP/Provisioner/internal/operations/ensure_posix_account.go: ########## @@ -0,0 +1,401 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package operations + +import ( + "context" + "fmt" + "log/slog" + "regexp" + "strconv" + + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/client" + "github.com/apache/airavata-custos/pkg/models" +) + +// identitySourcePrefix is the user_identities.source prefix for +// LDAP-assigned uidNumbers. The full source string is +// identitySourcePrefix + ":" + CustosClusterID so a deployment +// servicing multiple clusters caches each cluster's uidNumber +// separately — the same Custos user can have different POSIX uids +// on different clusters. Symmetric to the COmanage connector's +// source="comanage" tag but scoped per cluster to prevent the +// cross-cluster UID reuse that a flat source= would allow. +const identitySourcePrefix = "ldap" + +// identitySource returns the fully-qualified user_identities.source +// value for this connector instance's cluster. +func (o *Orchestrator) identitySource() string { + if id := o.c.Config().CustosClusterID; id != "" { + return identitySourcePrefix + ":" + id + } + return identitySourcePrefix +} + +// maxPosixUsernameLen is the traditional POSIX login-name cap. Matches +// what the pkg/posix allocator already enforces on the AMIE side. +const maxPosixUsernameLen = 32 + +// validPosixUsername matches a lowercase POSIX-conformant login name. +// Restricting to [a-z0-9_-] also happens to reject every RFC 4514 DN +// metacharacter (comma, plus, quote, backslash, less/greater, semicolon, +// equals, hash, space), so the RDN can be safely concatenated into +// uid=<value>,<BaseDN> without DN escaping. +var validPosixUsername = regexp.MustCompile(`^[a-z_][a-z0-9_-]*$`) + +// maxAllocRetries bounds how many times the orchestrator will re-allocate +// after a uidNumber constraint violation from an out-of-band writer that +// already holds our counter's next value. The persistent monotonic +// counter serialises Custos-side races itself (InnoDB row lock on the +// ldap_uid_sequence row), so retries are only reached against uids +// claimed by processes outside this connector. +const maxAllocRetries = 3 + +// ensurePOSIXAccountImpl provisions a posixAccount entry (and, when +// GroupBaseDN is configured, a matching posixGroup) for the given +// ComputeClusterUser. Mirrors the COmanage connector's shape: resolve +// the identity registry's assigned id, cache it in user_identities, +// materialise the POSIX record, and add the primary group. +// +// Direct-LDAP path specifics: +// - The identity registry is a persistent monotonic counter +// (internal/store.UIDSequence, one row per cluster in +// ldap_uid_sequence). Never regresses on entry deletion, so a new +// user cannot inherit a deleted user's numeric uid. +// - gidnumber = uidnumber (one-group-per-user), same pattern the +// COmanage flow uses on the CoGroup identifier. +// - user_identities(source="ldap:<clusterID>", external_id=<uidNumber>) +// caches the assignment so re-provisioning the same user is O(1). +// - When GroupBaseDN is set, a posixGroup with cn=<LocalUsername>, +// gidNumber=<uidNumber> is created too; without GroupBaseDN the +// connector assumes auto-private-groups on the client side. +func (o *Orchestrator) ensurePOSIXAccountImpl(ctx context.Context, cu *models.ComputeClusterUser) error { + log := slog.With( + "correlation_id", cu.ID, + "custos_user_id", cu.UserID, + "local_username", cu.LocalUsername, + "base_dn", o.c.Config().BaseDN, + ) + + if cu.LocalUsername == "" { + err := fmt.Errorf("compute_cluster_user %s has empty local_username", cu.ID) + o.dlq(ctx, cu, "validate_local_username", err) + return err + } + if len(cu.LocalUsername) > maxPosixUsernameLen || !validPosixUsername.MatchString(cu.LocalUsername) { + err := fmt.Errorf("compute_cluster_user %s has invalid local_username %q (must be POSIX-safe: lowercase, [a-z_][a-z0-9_-]*, <=%d chars)", + cu.ID, cu.LocalUsername, maxPosixUsernameLen) + o.dlq(ctx, cu, "validate_local_username", err) + return err + } + + user, err := o.core.GetUser(ctx, cu.UserID) + if err != nil { + o.dlq(ctx, cu, "get_custos_user", err) + return fmt.Errorf("get custos user %s: %w", cu.UserID, err) + } + if user == nil { + err := fmt.Errorf("custos user %s not found", cu.UserID) + o.dlq(ctx, cu, "get_custos_user", err) + return err + } + + // 1. Cache lookup — has this user already been assigned a uidNumber? + cached, err := o.findCachedUID(ctx, cu.UserID) + if err != nil { + o.dlq(ctx, cu, "list_user_identities", err) + return err + } + if cached > 0 { + log.Info("ldap provisioner: uidNumber cache hit", "uid", cached) + return o.ensureEntry(ctx, cu, user, cached) + } + + // 2. LDAP lookup — the entry may already exist (out-of-band or a + // prior run that failed to cache). Adopt its uidNumber. + adopted, err := o.tryAdoptExisting(ctx, cu, user) + if err != nil { + return err + } + if adopted { + return nil + } + + // 3. New user — pull a fresh uid from the persistent monotonic + // counter, then Add. The counter is a per-cluster row updated + // via one atomic UPDATE (InnoDB row lock), so allocations never + // regress even after LDAP entries are deleted and never collide + // with concurrent allocators. Constraint violations (from + // out-of-band writers or a mis-seeded counter) drive a bounded + // retry that pulls the next value. + cfg := o.c.Config() + for attempt := 1; attempt <= maxAllocRetries; attempt++ { + uid, allocErr := o.uids.Allocate(ctx, cfg.CustosClusterID) + if allocErr != nil { + o.dlq(ctx, cu, "allocate_uid", allocErr) + return allocErr + } + + acct := buildPosixAccount(cu, user, uid, cfg) + writtenDN, addErr := o.c.AddPosixAccount(acct) + if addErr == nil { + if cacheErr := o.storeUIDIdentity(ctx, cu.UserID, uid); cacheErr != nil { + log.Error("ldap provisioner: cache write failed after LDAP add; partial state, self-heals on next event", + "uid", uid, "err", cacheErr) + } + if err := o.ensurePrimaryGroup(ctx, cu, uid, log); err != nil { + o.dlq(ctx, cu, "ensure_primary_group", err) + return err + } + o.audit(ctx, cu, "LDAPAccountCreated", + fmt.Sprintf("dn=%s username=%s uid=%d gid=%d", writtenDN, cu.LocalUsername, uid, uid)) + log.Info("ldap provisioner: posixAccount created", "dn", writtenDN, "uid", uid) + return nil + } + if client.IsAlreadyExists(addErr) { + // Another Custos instance created the entry between our + // Find and Add. Adopt what's there rather than DLQ. + log.Info("ldap provisioner: entry created concurrently by another instance, adopting") + adopted, adoptErr := o.tryAdoptExisting(ctx, cu, user) + if adoptErr != nil { + return adoptErr + } + if adopted { + return nil + } + log.Warn("ldap provisioner: EntryAlreadyExists but Find empty, retrying", + "attempt", attempt) + continue + } + if !client.IsConstraintViolation(addErr) { + o.dlq(ctx, cu, "add_posix_account", addErr) + return addErr + } + // uidNumber uniqueness violation from an out-of-band writer + // (some entry in LDAP already has this number, e.g. a system + // user provisioned outside Custos). Pull the next value from + // the counter and try again. + log.Warn("ldap provisioner: uidNumber constraint violation, allocating next", + "attempt", attempt, "attempted_uid", uid) + } + + err = fmt.Errorf("failed to write posixAccount after %d retries", maxAllocRetries) + o.dlq(ctx, cu, "allocate_uid_retries_exhausted", err) + return err +} + +// tryAdoptExisting looks up a posixAccount by LocalUsername and, if +// present, caches its uidNumber and syncs mutable attributes via +// ensureEntry. Returns (adopted, err): +// +// - (true, nil) — an existing entry was adopted and fully processed +// (audit fired, group ensured); caller returns nil. +// - (false, nil) — no matching entry; caller falls through to +// fresh allocation. +// - (false, err) — an error occurred; DLQ has already been written. +// +// Used both by the initial LDAP-lookup path and by the retry loop when +// a concurrent instance's Add races ahead of ours. +func (o *Orchestrator) tryAdoptExisting(ctx context.Context, cu *models.ComputeClusterUser, user *models.User) (bool, error) { + dn, attrs, err := o.c.FindPosixAccount(cu.LocalUsername) Review Comment: Matches on `local_username` alone, so an existing entry for a different person gets adopted and then overwritten by `ModifyPosixAccount` at 296. With the suggested table, look up by `compute_cluster_user_id` instead. No row plus an existing entry with that username is a conflict, so DLQ it. Adopting an existing account should be an explicit import. ########## connectors/LDAP/Provisioner/pkg/ldap/loader.go: ########## @@ -0,0 +1,277 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package ldap is the LDAP Provisioner entry point. Wired from +// internal/connectors/loader.go alongside the COmanage connector — a site +// runs one or the other depending on whether it uses COmanage as a +// managed layer in front of the directory. +package ldap + +import ( + "context" + "log/slog" + "os" + "strconv" + "sync" + "time" + + "github.com/jmoiron/sqlx" + + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/client" + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/store" + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/subscribers" + ldapdb "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/db" + "github.com/apache/airavata-custos/internal/config" + "github.com/apache/airavata-custos/internal/db" + "github.com/apache/airavata-custos/internal/tracing" + "github.com/apache/airavata-custos/pkg/events" + "github.com/apache/airavata-custos/pkg/identity" + "github.com/apache/airavata-custos/pkg/service" +) + +const connectorName = "ldap" + +// init registers the event types that close out an audit trace for this +// connector, so the audit-trace UI marks a provisioning run as complete +// instead of leaving it at in_progress. Mirrors the equivalent block in +// the COmanage and AMIE loaders. +func init() { + tracing.RegisterTerminalMarkers("ldap", + "LDAPAccountCreated", + "LDAPAccountUpdated", + "LDAPGroupCreated", + ) +} + +// LoadConnector wires the subscriber to the event bus. Reads YAML config +// first and falls back to environment variables. If neither yields a +// complete config, it logs and returns nil without registering — same +// skip-with-log pattern the other connectors use so a dev server boots +// without LDAP credentials. +func LoadConnector(ctx context.Context, database *sqlx.DB, eventBus *events.Bus, coreService *service.Service, _ *sync.WaitGroup, _ *identity.Router, connectorConfig *config.ConnectorConfig) error { + cfg, ok := loadConfigFromConnectorConfig(connectorConfig) + if !ok { + cfg, ok = loadConfigFromEnv() + if !ok { + slog.Info("ldap provisioner: required config not set; skipping") + return nil + } + } + + if err := db.MigrateConnectorFS(database, ldapdb.MigrationFS(), "migrations", connectorName); err != nil { + return err + } + + ldapClient, err := client.New(cfg) + if err != nil { + return err + } + + uidSeq := store.NewUIDSequence(database) + if err := seedUIDCounter(ctx, ldapClient, uidSeq, cfg); err != nil { + return err + } + + subscribers.NewClusterUserSubscriber(ldapClient, uidSeq, eventBus, coreService, cfg.CustosClusterID).RegisterSubscribers() + slog.Info("ldap provisioner: subscriber registered", + "url", cfg.URL, "base_dn", cfg.BaseDN, "cluster_id", cfg.CustosClusterID) + // Custos is the source of truth for provisioning: entries created by this + // connector must not be edited directly in LDAP. There is no drift + // reconciliation; out-of-band edits will be invisible to Custos. + return nil +} + +// seedUIDCounter initialises the ldap_uid_sequence row for this +// cluster. Runs one LDAP scan to find max(existing uidNumber) so the +// counter starts above any entries provisioned out-of-band before +// this connector ran. Idempotent: on subsequent boots the store's +// GREATEST() upsert preserves whatever value the counter has grown +// to, so a re-scan cannot regress the sequence. +func seedUIDCounter(ctx context.Context, ldapClient *client.Client, seq *store.UIDSequence, cfg client.Config) error { + minUID := cfg.MinUID + if minUID <= 0 { + minUID = client.DefaultMinUID + } + + // One-time LDAP scan to find the current head of the uidNumber + // range. AllocateNextUID(minUID) returns max(uidNumber)+1, floored + // at minUID — exactly the value we want as the counter's next_uid. + // If the scan fails (e.g. LDAP unreachable at boot), fall back to + // the floor so the connector still starts. + initial, err := ldapClient.AllocateNextUID(minUID) + if err != nil { + slog.Warn("ldap provisioner: could not scan LDAP for existing max uidNumber during seed; falling back to floor", + "floor", minUID, "err", err) + initial = minUID + } + + if err := seq.Seed(ctx, cfg.CustosClusterID, initial); err != nil { + return err + } + slog.Info("ldap provisioner: uid sequence seeded", + "cluster_id", cfg.CustosClusterID, "initial_next_uid", initial) + return nil +} + +func loadConfigFromConnectorConfig(connectorConfig *config.ConnectorConfig) (client.Config, bool) { + if connectorConfig == nil { + return client.Config{}, false + } + + var url, bindDN, bindPassword, baseDN, custosCluster, defaultShell, homedirPrefix string + verifySSL := true + timeout := 30 * time.Second + minUID := client.DefaultMinUID + + var groupBaseDN string + + if directory, err := connectorConfig.GetNestedConfig("directory"); err == nil { + if v, ok := directory["url"].(string); ok { + url = v + } + if v, ok := directory["bind_dn"].(string); ok { + bindDN = v + } + if v, ok := directory["bind_password"].(string); ok { + bindPassword = v + } + if v, ok := directory["base_dn"].(string); ok { + baseDN = v + } + if v, ok := directory["group_base_dn"].(string); ok { + groupBaseDN = v + } + if v, ok := directory["verify_ssl"].(bool); ok { + verifySSL = v + } + } + + if provisioning, err := connectorConfig.GetNestedConfig("provisioning"); err == nil { + if v, ok := provisioning["custos_cluster_id"].(string); ok { + custosCluster = v + } + if v, ok := provisioning["default_shell"].(string); ok { + defaultShell = v + } + if v, ok := provisioning["homedir_prefix"].(string); ok { + homedirPrefix = v + } + if v, ok := provisioning["http_timeout"].(string); ok { Review Comment: `http_timeout` is the wrong name for LDAP. ########## connectors/LDAP/Provisioner/internal/subscribers/cluster_user.go: ########## @@ -0,0 +1,101 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package subscribers wires ComputeClusterUserCreateEvent to the LDAP +// orchestrator. Parallels the COmanage connector's subscriber; a +// deployment picks one or the other depending on whether it fronts LDAP +// with COmanage. +package subscribers + +import ( + "context" + "log/slog" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/client" + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/operations" + "github.com/apache/airavata-custos/internal/audit" + "github.com/apache/airavata-custos/internal/tracing" + "github.com/apache/airavata-custos/pkg/events" + "github.com/apache/airavata-custos/pkg/models" + "github.com/apache/airavata-custos/pkg/service" +) + +// ClusterUserSubscriber listens for ComputeClusterUserCreateEvent and +// drives the orchestrator. Events whose ComputeClusterID does not match +// CustosClusterID are dropped — a deployment servicing multiple clusters +// runs one subscriber instance per cluster. +type ClusterUserSubscriber struct { + ops *operations.Orchestrator + bus *events.Bus + core *service.Service + custosClusterID string +} + +func NewClusterUserSubscriber(c *client.Client, uids operations.UIDAllocator, bus *events.Bus, core *service.Service, custosClusterID string) *ClusterUserSubscriber { + return &ClusterUserSubscriber{ + ops: operations.New(c, core, uids), + bus: bus, + core: core, + custosClusterID: custosClusterID, + } +} + +func (s *ClusterUserSubscriber) RegisterSubscribers() { + s.bus.Subscribe(events.ComputeClusterUserCreateEvent, s.handleClusterUserCreate) Review Comment: Only the create event is handled, so deactivating a membership leaves the LDAP account live. ########## connectors/LDAP/Provisioner/internal/operations/ensure_posix_account.go: ########## @@ -0,0 +1,401 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package operations + +import ( + "context" + "fmt" + "log/slog" + "regexp" + "strconv" + + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/client" + "github.com/apache/airavata-custos/pkg/models" +) + +// identitySourcePrefix is the user_identities.source prefix for +// LDAP-assigned uidNumbers. The full source string is +// identitySourcePrefix + ":" + CustosClusterID so a deployment +// servicing multiple clusters caches each cluster's uidNumber +// separately — the same Custos user can have different POSIX uids +// on different clusters. Symmetric to the COmanage connector's +// source="comanage" tag but scoped per cluster to prevent the +// cross-cluster UID reuse that a flat source= would allow. +const identitySourcePrefix = "ldap" + +// identitySource returns the fully-qualified user_identities.source +// value for this connector instance's cluster. +func (o *Orchestrator) identitySource() string { + if id := o.c.Config().CustosClusterID; id != "" { + return identitySourcePrefix + ":" + id + } + return identitySourcePrefix +} + +// maxPosixUsernameLen is the traditional POSIX login-name cap. Matches +// what the pkg/posix allocator already enforces on the AMIE side. +const maxPosixUsernameLen = 32 + +// validPosixUsername matches a lowercase POSIX-conformant login name. +// Restricting to [a-z0-9_-] also happens to reject every RFC 4514 DN +// metacharacter (comma, plus, quote, backslash, less/greater, semicolon, +// equals, hash, space), so the RDN can be safely concatenated into +// uid=<value>,<BaseDN> without DN escaping. +var validPosixUsername = regexp.MustCompile(`^[a-z_][a-z0-9_-]*$`) + +// maxAllocRetries bounds how many times the orchestrator will re-allocate +// after a uidNumber constraint violation from an out-of-band writer that +// already holds our counter's next value. The persistent monotonic +// counter serialises Custos-side races itself (InnoDB row lock on the +// ldap_uid_sequence row), so retries are only reached against uids +// claimed by processes outside this connector. +const maxAllocRetries = 3 + +// ensurePOSIXAccountImpl provisions a posixAccount entry (and, when +// GroupBaseDN is configured, a matching posixGroup) for the given +// ComputeClusterUser. Mirrors the COmanage connector's shape: resolve +// the identity registry's assigned id, cache it in user_identities, +// materialise the POSIX record, and add the primary group. +// +// Direct-LDAP path specifics: +// - The identity registry is a persistent monotonic counter +// (internal/store.UIDSequence, one row per cluster in +// ldap_uid_sequence). Never regresses on entry deletion, so a new +// user cannot inherit a deleted user's numeric uid. +// - gidnumber = uidnumber (one-group-per-user), same pattern the +// COmanage flow uses on the CoGroup identifier. +// - user_identities(source="ldap:<clusterID>", external_id=<uidNumber>) +// caches the assignment so re-provisioning the same user is O(1). +// - When GroupBaseDN is set, a posixGroup with cn=<LocalUsername>, +// gidNumber=<uidNumber> is created too; without GroupBaseDN the +// connector assumes auto-private-groups on the client side. Review Comment: Trim these. Most of the block restates what the code does. (update the rest of the comments as well) ########## connectors/LDAP/Provisioner/db/migrations/000001_uid_sequence.up.sql: ########## @@ -0,0 +1,30 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- Persistent monotonic UID allocator per cluster. The row for a given +-- cluster_id holds the NEXT uidNumber to hand out; allocation is a +-- single UPDATE that InnoDB serialises via row locking, so concurrent +-- allocators (in-process or cross-process) cannot pick the same value. +-- +-- Rows are never deleted — decommissioning a cluster is handled at a +-- higher level. Never-decrementing counter is what guarantees no UID +-- reuse after an LDAP entry is deleted. +CREATE TABLE ldap_uid_sequence ( Review Comment: I think we should store: ``` ldap_posix_accounts compute_cluster_user_id uid_number gid_number username home_directory login_shell entry_dn status -> DEFAULT 'ACTIVE' created_at / updated_at UNIQUE (uid_number) UNIQUE (username) ``` Key it on `compute_cluster_users.id`, not `users.user_id`. That table is already unique per cluster and user, so we get the per cluster scoping for free and don't need the `ldap:<cluster_id>` tag. ########## connectors/LDAP/Provisioner/pkg/ldap/loader.go: ########## @@ -0,0 +1,277 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package ldap is the LDAP Provisioner entry point. Wired from +// internal/connectors/loader.go alongside the COmanage connector — a site +// runs one or the other depending on whether it uses COmanage as a +// managed layer in front of the directory. +package ldap + +import ( + "context" + "log/slog" + "os" + "strconv" + "sync" + "time" + + "github.com/jmoiron/sqlx" + + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/client" + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/store" + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/subscribers" + ldapdb "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/db" + "github.com/apache/airavata-custos/internal/config" + "github.com/apache/airavata-custos/internal/db" + "github.com/apache/airavata-custos/internal/tracing" + "github.com/apache/airavata-custos/pkg/events" + "github.com/apache/airavata-custos/pkg/identity" + "github.com/apache/airavata-custos/pkg/service" +) + +const connectorName = "ldap" + +// init registers the event types that close out an audit trace for this +// connector, so the audit-trace UI marks a provisioning run as complete +// instead of leaving it at in_progress. Mirrors the equivalent block in +// the COmanage and AMIE loaders. +func init() { + tracing.RegisterTerminalMarkers("ldap", + "LDAPAccountCreated", + "LDAPAccountUpdated", + "LDAPGroupCreated", + ) +} + +// LoadConnector wires the subscriber to the event bus. Reads YAML config +// first and falls back to environment variables. If neither yields a +// complete config, it logs and returns nil without registering — same +// skip-with-log pattern the other connectors use so a dev server boots +// without LDAP credentials. +func LoadConnector(ctx context.Context, database *sqlx.DB, eventBus *events.Bus, coreService *service.Service, _ *sync.WaitGroup, _ *identity.Router, connectorConfig *config.ConnectorConfig) error { + cfg, ok := loadConfigFromConnectorConfig(connectorConfig) + if !ok { + cfg, ok = loadConfigFromEnv() + if !ok { + slog.Info("ldap provisioner: required config not set; skipping") + return nil + } + } + + if err := db.MigrateConnectorFS(database, ldapdb.MigrationFS(), "migrations", connectorName); err != nil { + return err + } + + ldapClient, err := client.New(cfg) + if err != nil { + return err + } + + uidSeq := store.NewUIDSequence(database) + if err := seedUIDCounter(ctx, ldapClient, uidSeq, cfg); err != nil { + return err + } + + subscribers.NewClusterUserSubscriber(ldapClient, uidSeq, eventBus, coreService, cfg.CustosClusterID).RegisterSubscribers() + slog.Info("ldap provisioner: subscriber registered", + "url", cfg.URL, "base_dn", cfg.BaseDN, "cluster_id", cfg.CustosClusterID) + // Custos is the source of truth for provisioning: entries created by this + // connector must not be edited directly in LDAP. There is no drift + // reconciliation; out-of-band edits will be invisible to Custos. + return nil +} + +// seedUIDCounter initialises the ldap_uid_sequence row for this +// cluster. Runs one LDAP scan to find max(existing uidNumber) so the +// counter starts above any entries provisioned out-of-band before +// this connector ran. Idempotent: on subsequent boots the store's +// GREATEST() upsert preserves whatever value the counter has grown +// to, so a re-scan cannot regress the sequence. +func seedUIDCounter(ctx context.Context, ldapClient *client.Client, seq *store.UIDSequence, cfg client.Config) error { + minUID := cfg.MinUID + if minUID <= 0 { + minUID = client.DefaultMinUID + } + + // One-time LDAP scan to find the current head of the uidNumber + // range. AllocateNextUID(minUID) returns max(uidNumber)+1, floored + // at minUID — exactly the value we want as the counter's next_uid. + // If the scan fails (e.g. LDAP unreachable at boot), fall back to + // the floor so the connector still starts. + initial, err := ldapClient.AllocateNextUID(minUID) + if err != nil { + slog.Warn("ldap provisioner: could not scan LDAP for existing max uidNumber during seed; falling back to floor", + "floor", minUID, "err", err) + initial = minUID + } + + if err := seq.Seed(ctx, cfg.CustosClusterID, initial); err != nil { + return err + } + slog.Info("ldap provisioner: uid sequence seeded", + "cluster_id", cfg.CustosClusterID, "initial_next_uid", initial) + return nil +} + +func loadConfigFromConnectorConfig(connectorConfig *config.ConnectorConfig) (client.Config, bool) { + if connectorConfig == nil { + return client.Config{}, false + } + + var url, bindDN, bindPassword, baseDN, custosCluster, defaultShell, homedirPrefix string + verifySSL := true + timeout := 30 * time.Second + minUID := client.DefaultMinUID + + var groupBaseDN string + + if directory, err := connectorConfig.GetNestedConfig("directory"); err == nil { + if v, ok := directory["url"].(string); ok { + url = v + } + if v, ok := directory["bind_dn"].(string); ok { + bindDN = v + } + if v, ok := directory["bind_password"].(string); ok { + bindPassword = v + } + if v, ok := directory["base_dn"].(string); ok { + baseDN = v + } + if v, ok := directory["group_base_dn"].(string); ok { + groupBaseDN = v + } + if v, ok := directory["verify_ssl"].(bool); ok { + verifySSL = v + } + } + + if provisioning, err := connectorConfig.GetNestedConfig("provisioning"); err == nil { + if v, ok := provisioning["custos_cluster_id"].(string); ok { + custosCluster = v + } + if v, ok := provisioning["default_shell"].(string); ok { + defaultShell = v + } + if v, ok := provisioning["homedir_prefix"].(string); ok { + homedirPrefix = v + } + if v, ok := provisioning["http_timeout"].(string); ok { + if d, err := time.ParseDuration(v); err == nil { Review Comment: Parse errors are dropped silently here, better to add a log line ########## connectors/LDAP/Provisioner/internal/client/client.go: ########## @@ -0,0 +1,472 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package client is the LDAP protocol wrapper the LDAP Provisioner uses to +// read and write directory entries. It parallels the REST client the +// COmanage Identity-Provisioner uses; different wire protocol, same +// architectural role. +package client + +import ( + "crypto/tls" + "errors" + "fmt" + "net" + "strconv" + "sync" + "time" + + "github.com/go-ldap/ldap/v3" +) + +// ErrNotFound is returned when a search comes back empty. Mirrors the +// COmanage client's ErrNotFound so orchestration code can use errors.Is. +var ErrNotFound = errors.New("ldap: not found") + +// Config carries the connection parameters and the per-cluster identity +// the connector serves. CustosClusterID lets a subscriber filter events so +// a single deployment can host multiple provisioner instances side by side. +type Config struct { + URL string + BindDN string + BindPassword string + BaseDN string + VerifySSL bool + CustosClusterID string + DefaultShell string + HomedirPrefix string + Timeout time.Duration + + // MinUID is the lowest POSIX uidNumber the allocator will hand out. + // Sites usually reserve 0..999 for system, 1000..49999 for local + // users, and 50000+ for federated / auto-provisioned users. Default + // 50000 when unset. + MinUID int64 + + // GroupBaseDN is the container for posixGroup entries — e.g. + // "ou=groups,dc=example,dc=edu". When empty, the connector skips + // posixGroup creation entirely (fine on systems using automatic + // private groups; needed for strict SSSD setups). + GroupBaseDN string +} + +// DefaultMinUID is the value used when Config.MinUID is zero. Chosen to +// sit above the typical local-user range on RHEL / Debian derivatives. +const DefaultMinUID int64 = 50000 + +// Connection abstracts the go-ldap operations the client uses. Defined here +// so unit tests can substitute a fake without dialing a real server. +type Connection interface { + Bind(username, password string) error + Add(req *ldap.AddRequest) error + Modify(req *ldap.ModifyRequest) error + Search(req *ldap.SearchRequest) (*ldap.SearchResult, error) + Close() error +} + +// Dialer opens a Connection. Injected so tests can bypass real networking. +type Dialer interface { + Dial(url string, verifySSL bool, timeout time.Duration) (Connection, error) +} + +type defaultDialer struct{} + +func (defaultDialer) Dial(url string, verifySSL bool, timeout time.Duration) (Connection, error) { + opts := []ldap.DialOpt{ + ldap.DialWithTLSConfig(&tls.Config{InsecureSkipVerify: !verifySSL}), + } + if timeout > 0 { + opts = append(opts, ldap.DialWithDialer(&net.Dialer{Timeout: timeout})) + } + return ldap.DialURL(url, opts...) +} + +// PosixAccount is the subset of LDAP attributes the connector maintains +// for each user. UIDNumber and GIDNumber are required — schema-conformant +// posixAccount entries must have both. +type PosixAccount struct { + UID string + UIDNumber int64 + GIDNumber int64 + GivenName string + Surname string + Mail string + HomeDirectory string + LoginShell string +} + +// Client is a thin, connection-reusing wrapper over go-ldap. All public +// methods take Client.mu; helpers with the `Locked` suffix assume the +// caller already holds it. +type Client struct { + cfg Config + dialer Dialer + + mu sync.Mutex + conn Connection +} + +// New constructs a Client backed by the real go-ldap dialer. +func New(cfg Config) (*Client, error) { + if cfg.URL == "" || cfg.BindDN == "" || cfg.BaseDN == "" { + return nil, errors.New("client.New: Config requires URL, BindDN, and BaseDN") + } + return &Client{cfg: cfg, dialer: defaultDialer{}}, nil +} + +// NewWithDialer is used by tests to inject a fake dialer. +func NewWithDialer(cfg Config, d Dialer) *Client { + return &Client{cfg: cfg, dialer: d} +} + +// Config returns the config the client was built with. +func (c *Client) Config() Config { return c.cfg } + +// Close releases the underlying connection. +func (c *Client) Close() { + c.mu.Lock() + defer c.mu.Unlock() + c.closeConn() +} + +// FindPosixAccount searches for a posixAccount entry by UID and returns +// its DN plus a map of the retrieved attributes. Returns (empty, nil, nil) +// when no entry matches. +func (c *Client) FindPosixAccount(uid string) (string, map[string][]string, error) { + c.mu.Lock() + defer c.mu.Unlock() + return c.findPosixAccountLocked(uid) +} + +// AddPosixAccount creates a new posixAccount + inetOrgPerson entry. +// Callers supply the uidNumber (typically from internal/store.UIDSequence, +// which is the persistent monotonic allocator that guarantees no reuse +// after entry deletion and serialises cross-process races via InnoDB +// row locking). +func (c *Client) AddPosixAccount(a PosixAccount) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + return c.addPosixAccountLocked(a) +} + +// ModifyPosixAccount replaces the mutable attributes of an existing entry. +// The uid RDN and the numeric IDs are not modified — a change in either +// would be a different account. +func (c *Client) ModifyPosixAccount(dn string, a PosixAccount) error { + c.mu.Lock() + defer c.mu.Unlock() + return c.modifyPosixAccountLocked(dn, a) +} + +// AllocateNextUID scans BaseDN for all posixAccount entries, reads their +// uidNumber attribute, and returns max(uidNumber) + 1, floored at minUID. +// +// Used exclusively for one-time seeding of the persistent uid counter +// at connector startup — the loader calls this so a fresh deployment +// initialises above any entries already in LDAP from out-of-band +// provisioning. Steady-state allocations go through internal/store's +// UIDSequence, which is monotonic across restarts and never regresses +// when LDAP entries are deleted. +func (c *Client) AllocateNextUID(minUID int64) (int64, error) { + c.mu.Lock() + defer c.mu.Unlock() + return c.allocateNextUIDLocked(minUID) +} + +// FindPosixGroup searches for a posixGroup entry by cn under GroupBaseDN +// and returns its DN. Returns "" when no entry matches. Errors when +// GroupBaseDN is empty — callers should check the config first. +func (c *Client) FindPosixGroup(cn string) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.cfg.GroupBaseDN == "" { + return "", errors.New("FindPosixGroup: GroupBaseDN not configured") + } + + conn, err := c.ensureConn() + if err != nil { + return "", err + } + filter := fmt.Sprintf("(&(objectClass=posixGroup)(cn=%s))", ldap.EscapeFilter(cn)) + req := ldap.NewSearchRequest( + c.cfg.GroupBaseDN, + ldap.ScopeWholeSubtree, ldap.NeverDerefAliases, + 2, c.searchTimeoutSeconds(), false, + filter, + []string{"cn", "gidNumber"}, + nil, + ) + result, err := conn.Search(req) + if err != nil { + c.closeConn() + return "", fmt.Errorf("search posixGroup cn=%s: %w", cn, err) + } + if len(result.Entries) == 0 { + return "", nil + } + if len(result.Entries) > 1 { + return "", fmt.Errorf("search posixGroup cn=%s returned %d entries", cn, len(result.Entries)) + } + return result.Entries[0].DN, nil +} + +// AddPosixGroup creates a posixGroup entry at cn=<cn>,<GroupBaseDN>. +// Errors when GroupBaseDN is empty. gidNumber must be positive. +func (c *Client) AddPosixGroup(cn string, gidNumber int64) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.cfg.GroupBaseDN == "" { + return "", errors.New("AddPosixGroup: GroupBaseDN not configured") + } + if cn == "" { + return "", errors.New("AddPosixGroup: cn is required") + } + if gidNumber <= 0 { + return "", errors.New("AddPosixGroup: gidNumber must be a positive integer") + } + + conn, err := c.ensureConn() + if err != nil { + return "", err + } + + dn := fmt.Sprintf("cn=%s,%s", cn, c.cfg.GroupBaseDN) + req := ldap.NewAddRequest(dn, nil) + req.Attribute("objectClass", []string{"top", "posixGroup"}) + req.Attribute("cn", []string{cn}) + req.Attribute("gidNumber", []string{strconv.FormatInt(gidNumber, 10)}) + + if err := conn.Add(req); err != nil { + c.closeConn() + return "", fmt.Errorf("add posixGroup %s: %w", dn, err) + } + return dn, nil +} + +// IsConstraintViolation reports whether err is an LDAP constraint / +// value-already-exists error — the signal that a concurrent writer +// claimed the uidNumber we chose. +func IsConstraintViolation(err error) bool { + return ldap.IsErrorWithCode(err, ldap.LDAPResultConstraintViolation) || + ldap.IsErrorWithCode(err, ldap.LDAPResultAttributeOrValueExists) +} + +// IsAlreadyExists reports whether err is an "entry already exists" LDAP +// error — used by group creation to treat concurrent adds as idempotent. +func IsAlreadyExists(err error) bool { + return ldap.IsErrorWithCode(err, ldap.LDAPResultEntryAlreadyExists) +} + +// ---- private locked helpers ------------------------------------------ + +func (c *Client) findPosixAccountLocked(uid string) (string, map[string][]string, error) { + conn, err := c.ensureConn() + if err != nil { + return "", nil, err + } + filter := fmt.Sprintf("(&(objectClass=posixAccount)(uid=%s))", ldap.EscapeFilter(uid)) + req := ldap.NewSearchRequest( + c.cfg.BaseDN, + ldap.ScopeWholeSubtree, ldap.NeverDerefAliases, + 2, c.searchTimeoutSeconds(), false, + filter, + []string{"uid", "uidNumber", "gidNumber", "cn", "givenName", "sn", "mail", "homeDirectory", "loginShell"}, + nil, + ) + result, err := conn.Search(req) + if err != nil { + c.closeConn() + return "", nil, fmt.Errorf("search posixAccount uid=%s: %w", uid, err) + } + if len(result.Entries) == 0 { + return "", nil, nil + } + if len(result.Entries) > 1 { + return "", nil, fmt.Errorf("search posixAccount uid=%s returned %d entries", uid, len(result.Entries)) + } + entry := result.Entries[0] + attrs := make(map[string][]string, len(entry.Attributes)) + for _, a := range entry.Attributes { + attrs[a.Name] = a.Values + } + return entry.DN, attrs, nil +} + +func (c *Client) addPosixAccountLocked(a PosixAccount) (string, error) { + if err := validate(a); err != nil { + return "", err + } + conn, err := c.ensureConn() + if err != nil { + return "", err + } + + dn := fmt.Sprintf("uid=%s,%s", a.UID, c.cfg.BaseDN) + req := ldap.NewAddRequest(dn, nil) + req.Attribute("objectClass", []string{ + "top", "person", "organizationalPerson", "inetOrgPerson", "posixAccount", + }) + req.Attribute("uid", []string{a.UID}) + req.Attribute("cn", []string{fullName(a)}) + req.Attribute("sn", []string{a.Surname}) + if a.GivenName != "" { + req.Attribute("givenName", []string{a.GivenName}) + } + req.Attribute("uidNumber", []string{strconv.FormatInt(a.UIDNumber, 10)}) + req.Attribute("gidNumber", []string{strconv.FormatInt(a.GIDNumber, 10)}) + req.Attribute("homeDirectory", []string{a.HomeDirectory}) + if a.LoginShell != "" { + req.Attribute("loginShell", []string{a.LoginShell}) + } + if a.Mail != "" { + req.Attribute("mail", []string{a.Mail}) + } + + if err := conn.Add(req); err != nil { + c.closeConn() + return "", fmt.Errorf("add posixAccount %s: %w", dn, err) + } + return dn, nil +} + +func (c *Client) modifyPosixAccountLocked(dn string, a PosixAccount) error { + conn, err := c.ensureConn() + if err != nil { + return err + } + req := ldap.NewModifyRequest(dn, nil) + req.Replace("cn", []string{fullName(a)}) + req.Replace("sn", []string{a.Surname}) + if a.GivenName != "" { + req.Replace("givenName", []string{a.GivenName}) + } + req.Replace("homeDirectory", []string{a.HomeDirectory}) + if a.LoginShell != "" { + req.Replace("loginShell", []string{a.LoginShell}) + } + if a.Mail != "" { + req.Replace("mail", []string{a.Mail}) + } + if err := conn.Modify(req); err != nil { + c.closeConn() + return fmt.Errorf("modify posixAccount %s: %w", dn, err) + } + return nil +} + +func (c *Client) allocateNextUIDLocked(minUID int64) (int64, error) { + if minUID <= 0 { + minUID = DefaultMinUID + } + conn, err := c.ensureConn() + if err != nil { + return 0, err + } + req := ldap.NewSearchRequest( + c.cfg.BaseDN, + ldap.ScopeWholeSubtree, ldap.NeverDerefAliases, + 0, c.searchTimeoutSeconds(), false, + "(&(objectClass=posixAccount)(uidNumber=*))", + []string{"uidNumber"}, + nil, + ) + result, err := conn.Search(req) + if err != nil { + c.closeConn() + return 0, fmt.Errorf("search posixAccount for max uidNumber: %w", err) + } + + var max int64 + for _, entry := range result.Entries { + for _, v := range entry.GetAttributeValues("uidNumber") { + n, err := strconv.ParseInt(v, 10, 64) + if err != nil { + continue + } + if n > max { + max = n + } + } + } + next := max + 1 + if next < minUID { + next = minUID + } + return next, nil +} + +// searchTimeoutSeconds returns Config.Timeout as whole seconds for use +// as the LDAP SearchRequest TimeLimit field. Clamped to a minimum of 1 +// so a sub-second configured timeout does not become 0 (which the LDAP +// protocol interprets as "no time limit" — the opposite of the caller's +// intent). +func (c *Client) searchTimeoutSeconds() int { + n := int(c.cfg.Timeout.Seconds()) + if n < 1 { + return 1 + } + return n +} + +func (c *Client) ensureConn() (Connection, error) { + if c.conn != nil { + return c.conn, nil + } + conn, err := c.dialer.Dial(c.cfg.URL, c.cfg.VerifySSL, c.cfg.Timeout) + if err != nil { + return nil, fmt.Errorf("dial LDAP %s: %w", c.cfg.URL, err) + } + if err := conn.Bind(c.cfg.BindDN, c.cfg.BindPassword); err != nil { + _ = conn.Close() + return nil, fmt.Errorf("bind LDAP as %s: %w", c.cfg.BindDN, err) + } + c.conn = conn + return conn, nil +} + +func (c *Client) closeConn() { + if c.conn != nil { + _ = c.conn.Close() + c.conn = nil + } +} + +func validate(a PosixAccount) error { + if a.UID == "" { + return errors.New("PosixAccount: UID is required") + } + if a.UIDNumber <= 0 { Review Comment: Once we hold `uid` and `gid` ourselves, also check they're at or above `min_uid`. ########## connectors/LDAP/Provisioner/internal/operations/ensure_posix_account.go: ########## @@ -0,0 +1,401 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package operations + +import ( + "context" + "fmt" + "log/slog" + "regexp" + "strconv" + + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/client" + "github.com/apache/airavata-custos/pkg/models" +) + +// identitySourcePrefix is the user_identities.source prefix for +// LDAP-assigned uidNumbers. The full source string is +// identitySourcePrefix + ":" + CustosClusterID so a deployment +// servicing multiple clusters caches each cluster's uidNumber +// separately — the same Custos user can have different POSIX uids +// on different clusters. Symmetric to the COmanage connector's +// source="comanage" tag but scoped per cluster to prevent the +// cross-cluster UID reuse that a flat source= would allow. +const identitySourcePrefix = "ldap" + +// identitySource returns the fully-qualified user_identities.source +// value for this connector instance's cluster. +func (o *Orchestrator) identitySource() string { + if id := o.c.Config().CustosClusterID; id != "" { + return identitySourcePrefix + ":" + id + } + return identitySourcePrefix +} + +// maxPosixUsernameLen is the traditional POSIX login-name cap. Matches +// what the pkg/posix allocator already enforces on the AMIE side. +const maxPosixUsernameLen = 32 + +// validPosixUsername matches a lowercase POSIX-conformant login name. +// Restricting to [a-z0-9_-] also happens to reject every RFC 4514 DN +// metacharacter (comma, plus, quote, backslash, less/greater, semicolon, +// equals, hash, space), so the RDN can be safely concatenated into +// uid=<value>,<BaseDN> without DN escaping. +var validPosixUsername = regexp.MustCompile(`^[a-z_][a-z0-9_-]*$`) + +// maxAllocRetries bounds how many times the orchestrator will re-allocate +// after a uidNumber constraint violation from an out-of-band writer that +// already holds our counter's next value. The persistent monotonic +// counter serialises Custos-side races itself (InnoDB row lock on the +// ldap_uid_sequence row), so retries are only reached against uids +// claimed by processes outside this connector. +const maxAllocRetries = 3 + +// ensurePOSIXAccountImpl provisions a posixAccount entry (and, when +// GroupBaseDN is configured, a matching posixGroup) for the given +// ComputeClusterUser. Mirrors the COmanage connector's shape: resolve +// the identity registry's assigned id, cache it in user_identities, +// materialise the POSIX record, and add the primary group. +// +// Direct-LDAP path specifics: +// - The identity registry is a persistent monotonic counter +// (internal/store.UIDSequence, one row per cluster in +// ldap_uid_sequence). Never regresses on entry deletion, so a new +// user cannot inherit a deleted user's numeric uid. +// - gidnumber = uidnumber (one-group-per-user), same pattern the +// COmanage flow uses on the CoGroup identifier. +// - user_identities(source="ldap:<clusterID>", external_id=<uidNumber>) +// caches the assignment so re-provisioning the same user is O(1). +// - When GroupBaseDN is set, a posixGroup with cn=<LocalUsername>, +// gidNumber=<uidNumber> is created too; without GroupBaseDN the +// connector assumes auto-private-groups on the client side. +func (o *Orchestrator) ensurePOSIXAccountImpl(ctx context.Context, cu *models.ComputeClusterUser) error { + log := slog.With( + "correlation_id", cu.ID, + "custos_user_id", cu.UserID, + "local_username", cu.LocalUsername, + "base_dn", o.c.Config().BaseDN, + ) + + if cu.LocalUsername == "" { + err := fmt.Errorf("compute_cluster_user %s has empty local_username", cu.ID) + o.dlq(ctx, cu, "validate_local_username", err) + return err + } + if len(cu.LocalUsername) > maxPosixUsernameLen || !validPosixUsername.MatchString(cu.LocalUsername) { + err := fmt.Errorf("compute_cluster_user %s has invalid local_username %q (must be POSIX-safe: lowercase, [a-z_][a-z0-9_-]*, <=%d chars)", + cu.ID, cu.LocalUsername, maxPosixUsernameLen) + o.dlq(ctx, cu, "validate_local_username", err) + return err + } + + user, err := o.core.GetUser(ctx, cu.UserID) + if err != nil { + o.dlq(ctx, cu, "get_custos_user", err) + return fmt.Errorf("get custos user %s: %w", cu.UserID, err) + } + if user == nil { + err := fmt.Errorf("custos user %s not found", cu.UserID) + o.dlq(ctx, cu, "get_custos_user", err) + return err + } + + // 1. Cache lookup — has this user already been assigned a uidNumber? + cached, err := o.findCachedUID(ctx, cu.UserID) + if err != nil { + o.dlq(ctx, cu, "list_user_identities", err) + return err + } + if cached > 0 { + log.Info("ldap provisioner: uidNumber cache hit", "uid", cached) + return o.ensureEntry(ctx, cu, user, cached) + } + + // 2. LDAP lookup — the entry may already exist (out-of-band or a + // prior run that failed to cache). Adopt its uidNumber. + adopted, err := o.tryAdoptExisting(ctx, cu, user) + if err != nil { + return err + } + if adopted { + return nil + } + + // 3. New user — pull a fresh uid from the persistent monotonic + // counter, then Add. The counter is a per-cluster row updated + // via one atomic UPDATE (InnoDB row lock), so allocations never + // regress even after LDAP entries are deleted and never collide + // with concurrent allocators. Constraint violations (from + // out-of-band writers or a mis-seeded counter) drive a bounded + // retry that pulls the next value. + cfg := o.c.Config() + for attempt := 1; attempt <= maxAllocRetries; attempt++ { + uid, allocErr := o.uids.Allocate(ctx, cfg.CustosClusterID) + if allocErr != nil { + o.dlq(ctx, cu, "allocate_uid", allocErr) + return allocErr + } + + acct := buildPosixAccount(cu, user, uid, cfg) + writtenDN, addErr := o.c.AddPosixAccount(acct) + if addErr == nil { + if cacheErr := o.storeUIDIdentity(ctx, cu.UserID, uid); cacheErr != nil { + log.Error("ldap provisioner: cache write failed after LDAP add; partial state, self-heals on next event", + "uid", uid, "err", cacheErr) + } + if err := o.ensurePrimaryGroup(ctx, cu, uid, log); err != nil { + o.dlq(ctx, cu, "ensure_primary_group", err) + return err + } + o.audit(ctx, cu, "LDAPAccountCreated", + fmt.Sprintf("dn=%s username=%s uid=%d gid=%d", writtenDN, cu.LocalUsername, uid, uid)) + log.Info("ldap provisioner: posixAccount created", "dn", writtenDN, "uid", uid) + return nil + } + if client.IsAlreadyExists(addErr) { + // Another Custos instance created the entry between our + // Find and Add. Adopt what's there rather than DLQ. + log.Info("ldap provisioner: entry created concurrently by another instance, adopting") + adopted, adoptErr := o.tryAdoptExisting(ctx, cu, user) + if adoptErr != nil { + return adoptErr + } + if adopted { + return nil + } + log.Warn("ldap provisioner: EntryAlreadyExists but Find empty, retrying", + "attempt", attempt) + continue + } + if !client.IsConstraintViolation(addErr) { + o.dlq(ctx, cu, "add_posix_account", addErr) + return addErr + } + // uidNumber uniqueness violation from an out-of-band writer + // (some entry in LDAP already has this number, e.g. a system + // user provisioned outside Custos). Pull the next value from + // the counter and try again. + log.Warn("ldap provisioner: uidNumber constraint violation, allocating next", + "attempt", attempt, "attempted_uid", uid) + } + + err = fmt.Errorf("failed to write posixAccount after %d retries", maxAllocRetries) + o.dlq(ctx, cu, "allocate_uid_retries_exhausted", err) + return err +} + +// tryAdoptExisting looks up a posixAccount by LocalUsername and, if +// present, caches its uidNumber and syncs mutable attributes via +// ensureEntry. Returns (adopted, err): +// +// - (true, nil) — an existing entry was adopted and fully processed +// (audit fired, group ensured); caller returns nil. +// - (false, nil) — no matching entry; caller falls through to +// fresh allocation. +// - (false, err) — an error occurred; DLQ has already been written. +// +// Used both by the initial LDAP-lookup path and by the retry loop when +// a concurrent instance's Add races ahead of ours. +func (o *Orchestrator) tryAdoptExisting(ctx context.Context, cu *models.ComputeClusterUser, user *models.User) (bool, error) { + dn, attrs, err := o.c.FindPosixAccount(cu.LocalUsername) + if err != nil { + o.dlq(ctx, cu, "find_posix_account", err) + return false, err + } + if dn == "" { + return false, nil + } + uid, ok := parseUIDFromAttrs(attrs) + if !ok { + err := fmt.Errorf("existing entry %s has no parseable uidNumber", dn) + o.dlq(ctx, cu, "parse_existing_uid", err) + return false, err + } + if cacheErr := o.storeUIDIdentity(ctx, cu.UserID, uid); cacheErr != nil { + // Adopted an existing LDAP entry but failed to cache. Next + // event for this user will re-adopt — non-fatal, but not + // silent. Escalated from Warn. + slog.Error("ldap provisioner: cache write failed after adoption; will re-adopt on next event", Review Comment: `correlation_id` is dropped here, it's set at the top of the run. Use the scoped logger everywhere so a run can be traced end to end. (applicable for the rest as well) ########## connectors/LDAP/Provisioner/internal/operations/orchestrator.go: ########## @@ -0,0 +1,117 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package operations provisions POSIX identity directly in LDAP for a +// (User, UnixCluster) pair. Parallels the COmanage Identity-Provisioner's +// operations package; direct-to-LDAP path for sites that don't run a +// COmanage Registry. +package operations + +import ( + "context" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/client" + "github.com/apache/airavata-custos/internal/tracing" + "github.com/apache/airavata-custos/pkg/models" + "github.com/apache/airavata-custos/pkg/service" +) + +// CoreService is the subset of pkg/service.Service the orchestrator needs. +// Declared here so tests can substitute a fake without standing up a real DB. +type CoreService interface { + GetUser(ctx context.Context, id string) (*models.User, error) + ListUserIdentitiesForUser(ctx context.Context, userID string) ([]models.UserIdentity, error) + CreateUserIdentity(ctx context.Context, ui *models.UserIdentity) (*models.UserIdentity, error) + CreateAuditEvent(ctx context.Context, e *models.AuditEvent) (*models.AuditEvent, error) +} + +// UIDAllocator hands out fresh uidNumbers from a persistent, monotonic +// counter. Never regresses — deleting an LDAP entry does not free the +// number for reuse, which is what closes the "new user inherits deleted +// user's files" hole. Backed by internal/store.UIDSequence in +// production; interfaced here so tests can substitute an in-memory +// counter. +type UIDAllocator interface { + Allocate(ctx context.Context, clusterID string) (int64, error) +} + +// Orchestrator wraps the LDAP client, the core service, and the UID +// allocator. Exposes a single EnsurePOSIXAccount call that the +// subscriber invokes on every ComputeClusterUserCreateEvent it accepts. +type Orchestrator struct { + c *client.Client + core CoreService + uids UIDAllocator +} + +// New builds an Orchestrator. Takes *service.Service directly so the +// loader wiring stays symmetrical with the COmanage connector; the +// exported CoreService interface is what the orchestrator's methods +// actually depend on. +func New(c *client.Client, core *service.Service, uids UIDAllocator) *Orchestrator { + return &Orchestrator{c: c, core: core, uids: uids} +} + +// EnsurePOSIXAccount is the entry point invoked by the subscriber. It +// wraps the real flow in a tracing span so failures surface with the +// step name attached. +func (o *Orchestrator) EnsurePOSIXAccount(ctx context.Context, cu *models.ComputeClusterUser) error { + ctx, span := tracing.Start(ctx, "ldap.ensure_posix_account") + defer span.End() + span.SetAttributes( + attribute.String("ldap.cluster_user_id", cu.ID), + attribute.String("ldap.user_id", cu.UserID), + attribute.String("ldap.base_dn", o.c.Config().BaseDN), + ) + if err := o.ensurePOSIXAccountImpl(ctx, cu); err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + return err + } + return nil +} + +// audit is the success-side helper: emits an event to the core audit log +// tagged with the ComputeClusterUser row so downstream tooling can trace +// the provisioning history for that user. +func (o *Orchestrator) audit(ctx context.Context, cu *models.ComputeClusterUser, eventType, details string) { + _, _ = o.core.CreateAuditEvent(ctx, &models.AuditEvent{ + EventType: eventType, + EntityID: cu.ID, + EntityType: "compute_cluster_user", + Details: details, + }) +} + +// dlq (dead-letter) is the failure-side helper: emits a ProvisioningFailed +// event with the step name so operators can grep the audit table for +// stuck users without correlating logs. +func (o *Orchestrator) dlq(ctx context.Context, cu *models.ComputeClusterUser, step string, err error) { Review Comment: dlq isn't the right name, something like `auditFailure` would be better ########## connectors/LDAP/Provisioner/internal/operations/ensure_posix_account.go: ########## @@ -0,0 +1,401 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package operations + +import ( + "context" + "fmt" + "log/slog" + "regexp" + "strconv" + + "github.com/apache/airavata-custos/connectors/LDAP/Provisioner/internal/client" + "github.com/apache/airavata-custos/pkg/models" +) + +// identitySourcePrefix is the user_identities.source prefix for +// LDAP-assigned uidNumbers. The full source string is +// identitySourcePrefix + ":" + CustosClusterID so a deployment +// servicing multiple clusters caches each cluster's uidNumber +// separately — the same Custos user can have different POSIX uids +// on different clusters. Symmetric to the COmanage connector's +// source="comanage" tag but scoped per cluster to prevent the +// cross-cluster UID reuse that a flat source= would allow. +const identitySourcePrefix = "ldap" + +// identitySource returns the fully-qualified user_identities.source +// value for this connector instance's cluster. +func (o *Orchestrator) identitySource() string { + if id := o.c.Config().CustosClusterID; id != "" { + return identitySourcePrefix + ":" + id + } + return identitySourcePrefix +} + +// maxPosixUsernameLen is the traditional POSIX login-name cap. Matches +// what the pkg/posix allocator already enforces on the AMIE side. +const maxPosixUsernameLen = 32 + +// validPosixUsername matches a lowercase POSIX-conformant login name. +// Restricting to [a-z0-9_-] also happens to reject every RFC 4514 DN +// metacharacter (comma, plus, quote, backslash, less/greater, semicolon, +// equals, hash, space), so the RDN can be safely concatenated into +// uid=<value>,<BaseDN> without DN escaping. +var validPosixUsername = regexp.MustCompile(`^[a-z_][a-z0-9_-]*$`) + +// maxAllocRetries bounds how many times the orchestrator will re-allocate +// after a uidNumber constraint violation from an out-of-band writer that +// already holds our counter's next value. The persistent monotonic +// counter serialises Custos-side races itself (InnoDB row lock on the +// ldap_uid_sequence row), so retries are only reached against uids +// claimed by processes outside this connector. +const maxAllocRetries = 3 + +// ensurePOSIXAccountImpl provisions a posixAccount entry (and, when +// GroupBaseDN is configured, a matching posixGroup) for the given +// ComputeClusterUser. Mirrors the COmanage connector's shape: resolve +// the identity registry's assigned id, cache it in user_identities, +// materialise the POSIX record, and add the primary group. +// +// Direct-LDAP path specifics: +// - The identity registry is a persistent monotonic counter +// (internal/store.UIDSequence, one row per cluster in +// ldap_uid_sequence). Never regresses on entry deletion, so a new +// user cannot inherit a deleted user's numeric uid. +// - gidnumber = uidnumber (one-group-per-user), same pattern the +// COmanage flow uses on the CoGroup identifier. +// - user_identities(source="ldap:<clusterID>", external_id=<uidNumber>) +// caches the assignment so re-provisioning the same user is O(1). +// - When GroupBaseDN is set, a posixGroup with cn=<LocalUsername>, +// gidNumber=<uidNumber> is created too; without GroupBaseDN the +// connector assumes auto-private-groups on the client side. +func (o *Orchestrator) ensurePOSIXAccountImpl(ctx context.Context, cu *models.ComputeClusterUser) error { + log := slog.With( + "correlation_id", cu.ID, + "custos_user_id", cu.UserID, + "local_username", cu.LocalUsername, + "base_dn", o.c.Config().BaseDN, + ) + + if cu.LocalUsername == "" { + err := fmt.Errorf("compute_cluster_user %s has empty local_username", cu.ID) + o.dlq(ctx, cu, "validate_local_username", err) + return err + } + if len(cu.LocalUsername) > maxPosixUsernameLen || !validPosixUsername.MatchString(cu.LocalUsername) { + err := fmt.Errorf("compute_cluster_user %s has invalid local_username %q (must be POSIX-safe: lowercase, [a-z_][a-z0-9_-]*, <=%d chars)", + cu.ID, cu.LocalUsername, maxPosixUsernameLen) + o.dlq(ctx, cu, "validate_local_username", err) + return err + } + + user, err := o.core.GetUser(ctx, cu.UserID) + if err != nil { + o.dlq(ctx, cu, "get_custos_user", err) + return fmt.Errorf("get custos user %s: %w", cu.UserID, err) + } + if user == nil { + err := fmt.Errorf("custos user %s not found", cu.UserID) + o.dlq(ctx, cu, "get_custos_user", err) + return err + } + + // 1. Cache lookup — has this user already been assigned a uidNumber? + cached, err := o.findCachedUID(ctx, cu.UserID) + if err != nil { + o.dlq(ctx, cu, "list_user_identities", err) + return err + } + if cached > 0 { + log.Info("ldap provisioner: uidNumber cache hit", "uid", cached) + return o.ensureEntry(ctx, cu, user, cached) + } + + // 2. LDAP lookup — the entry may already exist (out-of-band or a + // prior run that failed to cache). Adopt its uidNumber. + adopted, err := o.tryAdoptExisting(ctx, cu, user) + if err != nil { + return err + } + if adopted { + return nil + } + + // 3. New user — pull a fresh uid from the persistent monotonic + // counter, then Add. The counter is a per-cluster row updated + // via one atomic UPDATE (InnoDB row lock), so allocations never + // regress even after LDAP entries are deleted and never collide + // with concurrent allocators. Constraint violations (from + // out-of-band writers or a mis-seeded counter) drive a bounded + // retry that pulls the next value. + cfg := o.c.Config() + for attempt := 1; attempt <= maxAllocRetries; attempt++ { + uid, allocErr := o.uids.Allocate(ctx, cfg.CustosClusterID) + if allocErr != nil { + o.dlq(ctx, cu, "allocate_uid", allocErr) + return allocErr + } + + acct := buildPosixAccount(cu, user, uid, cfg) + writtenDN, addErr := o.c.AddPosixAccount(acct) + if addErr == nil { + if cacheErr := o.storeUIDIdentity(ctx, cu.UserID, uid); cacheErr != nil { + log.Error("ldap provisioner: cache write failed after LDAP add; partial state, self-heals on next event", + "uid", uid, "err", cacheErr) + } + if err := o.ensurePrimaryGroup(ctx, cu, uid, log); err != nil { + o.dlq(ctx, cu, "ensure_primary_group", err) + return err + } + o.audit(ctx, cu, "LDAPAccountCreated", + fmt.Sprintf("dn=%s username=%s uid=%d gid=%d", writtenDN, cu.LocalUsername, uid, uid)) + log.Info("ldap provisioner: posixAccount created", "dn", writtenDN, "uid", uid) + return nil + } + if client.IsAlreadyExists(addErr) { + // Another Custos instance created the entry between our + // Find and Add. Adopt what's there rather than DLQ. + log.Info("ldap provisioner: entry created concurrently by another instance, adopting") + adopted, adoptErr := o.tryAdoptExisting(ctx, cu, user) + if adoptErr != nil { + return adoptErr + } + if adopted { + return nil + } + log.Warn("ldap provisioner: EntryAlreadyExists but Find empty, retrying", + "attempt", attempt) + continue + } + if !client.IsConstraintViolation(addErr) { + o.dlq(ctx, cu, "add_posix_account", addErr) + return addErr + } + // uidNumber uniqueness violation from an out-of-band writer + // (some entry in LDAP already has this number, e.g. a system + // user provisioned outside Custos). Pull the next value from + // the counter and try again. + log.Warn("ldap provisioner: uidNumber constraint violation, allocating next", + "attempt", attempt, "attempted_uid", uid) + } + + err = fmt.Errorf("failed to write posixAccount after %d retries", maxAllocRetries) + o.dlq(ctx, cu, "allocate_uid_retries_exhausted", err) + return err +} + +// tryAdoptExisting looks up a posixAccount by LocalUsername and, if +// present, caches its uidNumber and syncs mutable attributes via +// ensureEntry. Returns (adopted, err): +// +// - (true, nil) — an existing entry was adopted and fully processed +// (audit fired, group ensured); caller returns nil. +// - (false, nil) — no matching entry; caller falls through to +// fresh allocation. +// - (false, err) — an error occurred; DLQ has already been written. +// +// Used both by the initial LDAP-lookup path and by the retry loop when +// a concurrent instance's Add races ahead of ours. +func (o *Orchestrator) tryAdoptExisting(ctx context.Context, cu *models.ComputeClusterUser, user *models.User) (bool, error) { + dn, attrs, err := o.c.FindPosixAccount(cu.LocalUsername) + if err != nil { + o.dlq(ctx, cu, "find_posix_account", err) + return false, err + } + if dn == "" { + return false, nil + } + uid, ok := parseUIDFromAttrs(attrs) + if !ok { + err := fmt.Errorf("existing entry %s has no parseable uidNumber", dn) + o.dlq(ctx, cu, "parse_existing_uid", err) + return false, err + } + if cacheErr := o.storeUIDIdentity(ctx, cu.UserID, uid); cacheErr != nil { + // Adopted an existing LDAP entry but failed to cache. Next + // event for this user will re-adopt — non-fatal, but not + // silent. Escalated from Warn. + slog.Error("ldap provisioner: cache write failed after adoption; will re-adopt on next event", + "user_id", cu.UserID, "uid", uid, "err", cacheErr) + } + slog.Info("ldap provisioner: adopted existing entry", + "dn", dn, "uid", uid, "user_id", cu.UserID) + if err := o.ensureEntry(ctx, cu, user, uid); err != nil { + return false, err + } + return true, nil +} + +// ensureEntry writes the posixAccount with a known uidNumber (from cache +// or from a pre-existing LDAP entry) and ensures the matching posixGroup +// exists when configured. +func (o *Orchestrator) ensureEntry(ctx context.Context, cu *models.ComputeClusterUser, user *models.User, uid int64) error { + log := slog.With( + "correlation_id", cu.ID, + "custos_user_id", cu.UserID, + "local_username", cu.LocalUsername, + "uid", uid, + ) + acct := buildPosixAccount(cu, user, uid, o.c.Config()) + + dn, _, err := o.c.FindPosixAccount(cu.LocalUsername) + if err != nil { + o.dlq(ctx, cu, "find_posix_account", err) + return err + } + + if dn == "" { + writtenDN, err := o.c.AddPosixAccount(acct) + if err != nil { + if client.IsAlreadyExists(err) { + // Race: the entry was created between our Find and + // our Add (e.g., another Custos instance provisioning + // the same user concurrently). Adopt the existing + // entry rather than emit a spurious failure. + log.Info("ldap provisioner: entry created concurrently during ensureEntry, adopting") + adopted, adoptErr := o.tryAdoptExisting(ctx, cu, user) + if adoptErr != nil { + return adoptErr + } + if adopted { + return nil + } + // EntryAlreadyExists but re-Find is empty — the racing + // entry may have been deleted between the Add response + // and our re-Find. Surface a real error. + race := fmt.Errorf("ensureEntry: EntryAlreadyExists but re-Find empty for %q", cu.LocalUsername) + o.dlq(ctx, cu, "ensure_entry_race", race) + return race + } + o.dlq(ctx, cu, "add_posix_account", err) + return err + } + if err := o.ensurePrimaryGroup(ctx, cu, uid, log); err != nil { + o.dlq(ctx, cu, "ensure_primary_group", err) + return err + } + o.audit(ctx, cu, "LDAPAccountCreated", + fmt.Sprintf("dn=%s username=%s uid=%d gid=%d", writtenDN, cu.LocalUsername, uid, uid)) + return nil + } + if err := o.c.ModifyPosixAccount(dn, acct); err != nil { + o.dlq(ctx, cu, "modify_posix_account", err) + return err + } + if err := o.ensurePrimaryGroup(ctx, cu, uid, log); err != nil { + o.dlq(ctx, cu, "ensure_primary_group", err) + return err + } + o.audit(ctx, cu, "LDAPAccountUpdated", + fmt.Sprintf("dn=%s username=%s", dn, cu.LocalUsername)) + return nil +} + +// ensurePrimaryGroup adds a posixGroup entry with cn=<local_username> +// and gidNumber=<uid> under GroupBaseDN, if that container is +// configured. Concurrent adds are tolerated: an "entry already exists" +// response is treated as success. +// +// When GroupBaseDN is empty the connector assumes the site uses +// automatic-private-groups on the client (RHEL / Fedora default) and +// no LDAP-side group entry is needed. +func (o *Orchestrator) ensurePrimaryGroup(ctx context.Context, cu *models.ComputeClusterUser, uid int64, log *slog.Logger) error { + if o.c.Config().GroupBaseDN == "" { + return nil + } + existing, err := o.c.FindPosixGroup(cu.LocalUsername) + if err != nil { + return fmt.Errorf("find posixGroup %s: %w", cu.LocalUsername, err) + } + if existing != "" { + log.Info("ldap provisioner: primary posixGroup already present", "dn", existing) + return nil + } + dn, err := o.c.AddPosixGroup(cu.LocalUsername, uid) + if err != nil { + if client.IsAlreadyExists(err) { + // A concurrent writer created it between our Find and Add. + log.Info("ldap provisioner: posixGroup created concurrently") + return nil + } + return fmt.Errorf("add posixGroup %s: %w", cu.LocalUsername, err) + } + log.Info("ldap provisioner: primary posixGroup created", "dn", dn, "gid", uid) + o.audit(ctx, cu, "LDAPGroupCreated", + fmt.Sprintf("dn=%s cn=%s gid=%d", dn, cu.LocalUsername, uid)) + return nil +} + +func buildPosixAccount(cu *models.ComputeClusterUser, user *models.User, uid int64, cfg client.Config) client.PosixAccount { + return client.PosixAccount{ + UID: cu.LocalUsername, + UIDNumber: uid, + GIDNumber: uid, // one-group-per-user, mirrors COmanage + GivenName: user.FirstName, + Surname: user.LastName, + Mail: user.Email, + HomeDirectory: cfg.HomedirPrefix + cu.LocalUsername, Review Comment: No separator here, so a prefix of "/home" gives "/homejsmith" -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
