Read groups from the groups service, and fix nested-group expansion

README.md

@@ -1,4 +1,4 @@
 group-propagator
 ================
 
-This service uses REST facades to Grouper and iRODS to copy group memberships from the former to the latter.
+This service uses REST facades to the DE groups service and iRODS to copy group memberships from the former to the latter.

client/datainfo/client.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"time"
 
 	"github.com/cyverse-de/go-mod/restutils"
 	"github.com/cyverse-de/group-propagator/logging"
@@ -27,7 +28,12 @@ type DataInfoClient struct {
 	DataInfoUser string
 }
 
-var httpClient = http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)}
+// The timeout keeps an unresponsive data-info from hanging the single AMQP
+// consumer goroutine indefinitely.
+var httpClient = http.Client{
+	Transport: otelhttp.NewTransport(http.DefaultTransport),
+	Timeout:   30 * time.Second,
+}
 
 func NewDataInfoClient(base, user string) *DataInfoClient {
 	return &DataInfoClient{base, user}
@@ -58,7 +64,7 @@ func (d *DataInfoClient) reqJSON(ctx context.Context, method, uri string, body i
 	if err != nil {
 		return errors.Wrap(err, "Failed requesting URL")
 	}
-	defer resp.Body.Close()
+	defer func() { _ = resp.Body.Close() }()
 
 	if resp.StatusCode < 200 || resp.StatusCode > 299 {
 		var e ServiceError

client/groups/client.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"time"
 
 	"github.com/cyverse-de/go-mod/restutils"
 	"github.com/cyverse-de/group-propagator/logging"
@@ -27,41 +28,40 @@ type GroupsClient struct {
 	GroupsID         string
 }
 
-var httpClient = http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)}
+// The timeout keeps an unresponsive groups service from hanging the single
+// AMQP consumer goroutine indefinitely.
+var httpClient = http.Client{
+	Transport: otelhttp.NewTransport(http.DefaultTransport),
+	Timeout:   30 * time.Second,
+}
 
 func NewGroupsClient(base string, user string, name string) *GroupsClient {
 	return &GroupsClient{GroupsBase: base, GroupsUser: user, DEUsersGroupName: name}
 }
 
-func (c *GroupsClient) getDEUsersGroupID(ctx context.Context) (*group, error) {
-	ctx, span := otel.Tracer(otelName).Start(ctx, "getGroupID")
+func (c *GroupsClient) getDEUsersGroup(ctx context.Context) (*Group, error) {
+	ctx, span := otel.Tracer(otelName).Start(ctx, "getDEUsersGroup")
 	defer span.End()
 
-	fullURL, err := url.Parse(c.GroupsBase)
+	uri, err := c.uriPath(ctx, fmt.Sprintf("group_type=%s&name=%s", GroupTypeSystem, url.QueryEscape(c.DEUsersGroupName)),
+		"groups", "lookup")
 	if err != nil {
-		return nil, errors.Wrap(err, "Failed to parse iplant-groups base URL")
+		return nil, errors.Wrap(err, "Failed to build the de-users lookup URL")
 	}
-	fullURL = fullURL.JoinPath("groups", c.DEUsersGroupName)
-
-	q := fullURL.Query()
-	q.Set("user", c.GroupsUser)
-
-	fullURL.RawQuery = q.Encode()
 
-	var group group
-	err = c.getJSON(ctx, fullURL.String(), &group)
-	if err != nil {
+	var g Group
+	if err := c.getJSON(ctx, uri, &g); err != nil {
 		return nil, errors.Wrap(err, "Failed to get group ID")
 	}
-	return &group, nil
+	return &g, nil
 }
 
 func (c *GroupsClient) SetGroupsID(ctx context.Context) error {
-	groups, err := c.getDEUsersGroupID(ctx)
+	g, err := c.getDEUsersGroup(ctx)
 	if err != nil {
 		return err
 	}
-	c.GroupsID = *groups.ID
+	c.GroupsID = g.ID
 	return nil
 }
 
@@ -89,10 +89,11 @@ func (c *GroupsClient) getJSON(ctx context.Context, uri string, target any) erro
 	resp, err := httpClient.Do(req)
 	if err != nil {
 		return errors.Wrap(err, "Failed requesting URL")
-	} else if resp.StatusCode < 200 || resp.StatusCode > 299 {
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode < 200 || resp.StatusCode > 299 {
 		return restutils.NewHTTPError(resp.StatusCode, fmt.Sprintf("GET %s returned %d", uri, resp.StatusCode))
 	}
-	defer resp.Body.Close()
 
 	if target != nil {
 		err = json.NewDecoder(resp.Body).Decode(target)
@@ -103,56 +104,121 @@ func (c *GroupsClient) getJSON(ctx context.Context, uri string, target any) erro
 	return err
 }
 
-// Use status endpoint to check our iplant-groups URI
+// Check reports whether the groups service is reachable and able to serve group
+// data. The service answers its status endpoint even when its database is down,
+// so a 200 alone is not enough to start against.
 func (c *GroupsClient) Check(ctx context.Context) error {
 	uri, err := url.Parse(c.GroupsBase)
 	if err != nil {
-		return errors.Wrap(err, "Failed to parse iplant-groups base URL")
+		return errors.Wrap(err, "Failed to parse groups base URL")
 	}
 
-	uri.RawQuery = "expecting=iplant-groups"
-
-	return c.getJSON(ctx, uri.String(), nil)
+	var status Status
+	if err := c.getJSON(ctx, uri.String(), &status); err != nil {
+		return err
+	}
+	if !status.Database {
+		return errors.New("the groups service cannot reach its database")
+	}
+	return nil
 }
 
-// List groups under a provided prefix, using the REST service
-func (c *GroupsClient) ListGroupsByPrefix(ctx context.Context, prefix, folder string) (GroupList, error) {
-	ctx, span := otel.Tracer(otelName).Start(ctx, "ListGroupsByPrefix")
-	defer span.End()
+// pageSize is how many groups one listing request asks for. The service caps a
+// single response at 1000, so anything larger would be silently trimmed.
+const pageSize = 1000
+
+// maxListPages bounds a listing crawl. At pageSize per page this covers orders
+// of magnitude more groups than a deployment holds, so reaching it means the
+// crawl is not making progress rather than that the deployment is large.
+const maxListPages = 1000
+
+// listGroups follows a group listing to its end, restricted to one group type
+// when groupType is non-empty.
+func (c *GroupsClient) listGroups(ctx context.Context, groupType string) ([]Group, error) {
+	// The offset advances by what each page actually held, and only an empty
+	// page ends the crawl. Keying either off the requested pageSize would
+	// silently truncate the listing if the service ever clamps responses
+	// below what we asked for.
+	var all []Group
+	seen := make(map[string]struct{})
+	for pages := 0; ; pages++ {
+		// A listing that never empties and never contributes a new group would
+		// otherwise spin forever, and with the AMQP consumer's concurrency of 1
+		// that halts all propagation while the accumulated slice grows.
+		if pages >= maxListPages {
+			return all, errors.Errorf(
+				"the group listing did not end after %d pages (%d groups); this usually means the "+
+					"groups service is ignoring the offset parameter", maxListPages, len(all))
+		}
 
-	var gs GroupList
-	var uri string
-	var err error
+		query := fmt.Sprintf("limit=%d&offset=%d", pageSize, len(all))
+		if groupType != "" {
+			query = fmt.Sprintf("group_type=%s&%s", url.QueryEscape(groupType), query)
+		}
+		uri, err := c.uriPath(ctx, query, "groups")
+		if err != nil {
+			return all, err
+		}
+		log.Debugf("listGroups uri: %s", uri)
 
-	if folder != "" {
-		uri, err = c.uriPath(ctx, fmt.Sprintf("search=%s&folder=%s", prefix, folder), "groups")
-	} else {
-		uri, err = c.uriPath(ctx, fmt.Sprintf("search=%s", prefix), "groups")
-	}
-	log.Debugf("ListGroupsByPrefix uri: %s", uri)
+		var page GroupList
+		if err := c.getJSON(ctx, uri, &page); err != nil {
+			return all, err
+		}
+		if len(page.Groups) == 0 {
+			return all, nil
+		}
 
-	if err != nil {
-		return gs, err
+		fresh := 0
+		for _, g := range page.Groups {
+			if _, dup := seen[g.ID]; dup {
+				continue
+			}
+			seen[g.ID] = struct{}{}
+			fresh++
+		}
+		if fresh == 0 {
+			return all, errors.Errorf(
+				"the group listing repeated a page of %d groups at offset %d; this usually means the "+
+					"groups service is ignoring the offset parameter", len(page.Groups), len(all))
+		}
+		all = append(all, page.Groups...)
 	}
-
-	err = c.getJSON(ctx, uri, &gs)
-	return gs, err
 }
 
-// Get the basic group information for a group from the REST service, given a name
-func (c *GroupsClient) GetGroupByName(ctx context.Context, groupName string) (Group, error) {
-	ctx, span := otel.Tracer(otelName).Start(ctx, "GetGroupByName")
+// ListAllGroups returns every group the service knows about, following
+// pagination to the end. The groups service is already scoped to one
+// deployment's group data, so there is no folder or prefix to filter on.
+func (c *GroupsClient) ListAllGroups(ctx context.Context) ([]Group, error) {
+	ctx, span := otel.Tracer(otelName).Start(ctx, "ListAllGroups")
 	defer span.End()
 
-	var g Group
+	return c.listGroups(ctx, "")
+}
 
-	uri, err := c.uriPath(ctx, "", "groups", url.PathEscape(groupName))
+// VerifyAdminListing proves the configured user gets unfiltered group
+// listings by checking that the de-users group appears in a system-type
+// listing; SetGroupsID must have run first. The check exists because the
+// service answers a non-admin with a 200 and an access-filtered page that
+// carries no marker -- without it, groups would silently stop propagating.
+func (c *GroupsClient) VerifyAdminListing(ctx context.Context) error {
+	ctx, span := otel.Tracer(otelName).Start(ctx, "VerifyAdminListing")
+	defer span.End()
+
+	system, err := c.listGroups(ctx, GroupTypeSystem)
 	if err != nil {
-		return g, err
+		return errors.Wrap(err, "Failed listing system groups")
 	}
-
-	err = c.getJSON(ctx, uri, &g)
-	return g, err
+	for _, g := range system {
+		if g.ID == c.GroupsID {
+			return nil
+		}
+	}
+	return errors.Errorf(
+		"the system group listing does not include %s (%s); this usually means the configured "+
+			"groups user %q is not in the groups service's admin-users list, so listings are "+
+			"access-filtered and propagation would silently miss groups",
+		c.DEUsersGroupName, c.GroupsID, c.GroupsUser)
 }
 
 // Get the basic group information for a group from the REST service, given an ID
@@ -162,7 +228,7 @@ func (c *GroupsClient) GetGroupByID(ctx context.Context, groupID string) (Group,
 
 	var g Group
 
-	uri, err := c.uriPath(ctx, "", "groups", "id", groupID)
+	uri, err := c.uriPath(ctx, "", "groups", url.PathEscape(groupID))
 	if err != nil {
 		return g, err
 	}
@@ -171,17 +237,76 @@ func (c *GroupsClient) GetGroupByID(ctx context.Context, groupID string) (Group,
 	return g, err
 }
 
-// List members of a group using the REST service, given a name
-func (c *GroupsClient) GetGroupMembers(ctx context.Context, groupName string) (GroupMembers, error) {
-	ctx, span := otel.Tracer(otelName).Start(ctx, "GetGroupMembers")
+// List members of a group using the REST service, given an ID. Membership is
+// keyed by ID rather than name because a nested group's member entry carries
+// its own short name, which is not a lookup key on either backend.
+func (c *GroupsClient) GetGroupMembersByID(ctx context.Context, groupID string) (GroupMembers, error) {
+	ctx, span := otel.Tracer(otelName).Start(ctx, "GetGroupMembersByID")
 	defer span.End()
 
+	// Always paged, never asked for unbounded: the service refuses an unpaged
+	// listing of a group larger than its own cap rather than truncating it, and
+	// propagation needs the whole membership -- the data-info update replaces
+	// rather than merges, so a short list removes people from the iRODS group.
+	//
+	// Termination mirrors listGroups: the offset advances by what the pages
+	// actually held and a page contributing no new member ends the crawl.
+	// Stopping on a page shorter than pageSize would truncate silently against a
+	// service whose own cap is lower than what we ask for.
 	var gm GroupMembers
-	uri, err := c.uriPath(ctx, "", "groups", url.PathEscape(groupName), "members")
-	if err != nil {
-		return gm, err
+	seen := make(map[string]struct{})
+	for pages := 0; ; pages++ {
+		if pages >= maxListPages {
+			return gm, errors.Errorf(
+				"the member listing for group %s did not end after %d pages (%d members); this "+
+					"usually means the groups service is ignoring the offset parameter",
+				groupID, maxListPages, len(gm.Members))
+		}
+
+		query := fmt.Sprintf("limit=%d&offset=%d", pageSize, len(gm.Members))
+		uri, err := c.uriPath(ctx, query, "groups", url.PathEscape(groupID), "members")
+		if err != nil {
+			return gm, err
+		}
+
+		var page GroupMembers
+		if err := c.getJSON(ctx, uri, &page); err != nil {
+			return gm, err
+		}
+		// Redaction describes the group, not the page: the membership was
+		// withheld rather than empty, and the caller must not read it as empty.
+		if page.Redacted {
+			return GroupMembers{Redacted: true}, nil
+		}
+		gm.Total = page.Total
+
+		fresh := 0
+		for _, m := range page.Members {
+			if _, dup := seen[m.ID]; dup {
+				continue
+			}
+			seen[m.ID] = struct{}{}
+			gm.Members = append(gm.Members, m)
+			fresh++
+		}
+		if fresh == 0 {
+			break
+		}
+		// The reported total ends the crawl without a further request that would
+		// only come back empty. Under-reporting it cannot truncate the result:
+		// the check below fails when the two disagree in either direction.
+		if page.Total > 0 && len(gm.Members) >= page.Total {
+			break
+		}
 	}
 
-	err = c.getJSON(ctx, uri, &gm)
-	return gm, err
+	// The service reports the group's whole membership alongside each page, so a
+	// crawl ending short of it assembled an incomplete list -- which must not
+	// reach data-info as though it were the whole one.
+	if gm.Total > 0 && len(gm.Members) != gm.Total {
+		return gm, errors.Errorf(
+			"collected %d of the %d members of group %s; refusing to propagate an incomplete membership",
+			len(gm.Members), gm.Total, groupID)
+	}
+	return gm, nil
 }