mirror of
https://github.com/gravitational/teleport.git
synced 2026-09-17 06:18:20 +08:00
tctl-investigate[1]: add activity-log search command (#67219)
* feat: add `tctl investigate` command Add a tctl subcommand that searches Teleport's Identity Security activity logs. The command exposes structured filter flags mirroring the Identity Security UI: ~19 include/exclude pairs mapping to Lucene fields (user, resource, IP, country, etc.). Multiple values on the same flag are OR'd; different flags are AND'd. `--query` accepts a raw Lucene expression for cases the structured flags can't express (wildcards, regex, OR across fields) and is mutually exclusive with the structured filters. Facets and events are fetched in parallel via errgroup. `total` is derived from the `event_type` facet on the stats endpoint, so it can drift a few percent from `len(data)` on long windows. This PR ships JSON/YAML output only. Text rendering, the geo filter, and the `--skill` flag land in the next PR in the stack. * fix: address codex suggestions * fix: update flag description
This commit is contained in:
Ghassan Gedeon Achi
@@ -1319,6 +1319,68 @@ Flags:
|
||||
|`--format`|`text` (one of: `text`, `json`, `yaml`)|Output format.|
|
||||
|`--[no-]connected`|`false`|Show locally connected instances summary|
|
||||
|
||||
## tctl investigate
|
||||
|
||||
Search and explore Identity Security activity logs.
|
||||
|
||||
Usage:
|
||||
|
||||
```code
|
||||
$ tctl investigate [<flags>]
|
||||
```
|
||||
|
||||
Flags:
|
||||
|
||||
|Flag|Default|Description|
|
||||
|---|---|---|
|
||||
|`--aws-account-id`|*no default*|Filter by AWS account ID (repeatable).|
|
||||
|`--aws-service`|*no default*|Filter by AWS service name (repeatable).|
|
||||
|`--city`|*no default*|Filter by city of origin (repeatable).|
|
||||
|`--country`|*no default*|Filter by country of origin (repeatable).|
|
||||
|`--event-type`|*no default*|Filter by event type, e.g. session.start (repeatable).|
|
||||
|`--exclude-aws-account-id`|*no default*|Exclude aws_account_id values (repeatable).|
|
||||
|`--exclude-aws-service`|*no default*|Exclude aws_service values (repeatable).|
|
||||
|`--exclude-city`|*no default*|Exclude city values (repeatable).|
|
||||
|`--exclude-country`|*no default*|Exclude country values (repeatable).|
|
||||
|`--exclude-event-type`|*no default*|Exclude event_type values (repeatable).|
|
||||
|`--exclude-github-org`|*no default*|Exclude github_organization values (repeatable).|
|
||||
|`--exclude-github-repo`|*no default*|Exclude github_repo values (repeatable).|
|
||||
|`--exclude-ip`|*no default*|Exclude ip values (repeatable).|
|
||||
|`--exclude-okta-org`|*no default*|Exclude okta_org values (repeatable).|
|
||||
|`--exclude-region`|*no default*|Exclude region values (repeatable).|
|
||||
|`--exclude-resource`|*no default*|Exclude target_resource values (repeatable).|
|
||||
|`--exclude-resource-kind`|*no default*|Exclude target_kind values (repeatable).|
|
||||
|`--exclude-source`|*no default*|Exclude event_source values (repeatable).|
|
||||
|`--exclude-status`|*no default* (any of (repeatable): `success`, `failure`)|Exclude status values (repeatable).|
|
||||
|`--exclude-teleport-cluster`|*no default*|Exclude teleport_cluster values (repeatable).|
|
||||
|`--exclude-token`|*no default*|Exclude token values (repeatable).|
|
||||
|`--exclude-user`|*no default*|Exclude identity_id values (repeatable).|
|
||||
|`--exclude-user-agent`|*no default*|Exclude user_agent values (repeatable).|
|
||||
|`--exclude-user-kind`|*no default*|Exclude identity_kind values (repeatable).|
|
||||
|`--format`|`yaml` (one of: `json`, `yaml`)|Output format. (Values: json, yaml)|
|
||||
|`--from`|`1d`|Include activity at or after this time. (Examples: 2006-01-02T15:04:05Z07:00, 2006-01-02, 24h, 7d; negative durations like -1h are future-relative. Default: 1d)|
|
||||
|`--github-org`|*no default*|Filter by GitHub organization (repeatable).|
|
||||
|`--github-repo`|*no default*|Filter by GitHub repository (repeatable).|
|
||||
|`--ip`|*no default*|Filter by source IP address (repeatable).|
|
||||
|`--limit`|`100`|Maximum number of events to return (0 for unlimited).|
|
||||
|`--[no-]facets-only`|`false`|Skip fetching events; return only the facet summary. Useful for narrowing a query before pulling logs.|
|
||||
|`--[no-]print-query`|`false`|Print the constructed query and exit without contacting the backend.|
|
||||
|`--[no-]show-unmatched`|`false`|Include facet values that exist in the time window but did not match the current filter (the backend reports these with count=-1). Useful for discovering filters to broaden.|
|
||||
|`--okta-org`|*no default*|Filter by Okta organization (repeatable).|
|
||||
|`--order`|`desc` (one of: `asc`, `desc`)|Result order by timestamp. (Values: asc, desc)|
|
||||
|`--query`|*no default*|Raw Lucene query. Mutually exclusive with structured filter flags. Example: --query 'identity_id:"alice@example.com" AND NOT status:"failure"'|
|
||||
|`--region`|*no default*|Filter by region (e.g. us-east-1 or a US state code; repeatable).|
|
||||
|`--resource`|*no default*|Filter by target resource (repeatable).|
|
||||
|`--resource-kind`|*no default*|Filter by resource kind, e.g. ssh, kube, session_recording (repeatable).|
|
||||
|`--source`|*no default*|Filter by event source (repeatable).|
|
||||
|`--status`|*no default* (any of (repeatable): `success`, `failure`)|Filter by event status (repeatable).|
|
||||
|`--teleport-cluster`|*no default*|Filter by Teleport cluster name (repeatable).|
|
||||
|`--to`|`now`|Include activity at or before this time. (Examples: 2006-01-02T15:04:05Z07:00, 2006-01-02, 24h, 7d; negative durations like -1h are future-relative. Default: now)|
|
||||
|`--token`|*no default*|Filter by token identifier (repeatable).|
|
||||
|`--user`|*no default*|Filter by user (email for users, ID for bots; repeatable).|
|
||||
|`--user-agent`|*no default*|Filter by user agent string (repeatable). Not populated on every Teleport event — use deliberately.|
|
||||
|`--user-kind`|*no default*|Filter by user kind, e.g. user, system (repeatable).|
|
||||
|
||||
## tctl kube ls
|
||||
|
||||
List all Kubernetes clusters registered with the cluster.
|
||||
|
||||
@@ -40,7 +40,8 @@ type AccessGraphCommand struct {
|
||||
config *servicecfg.Config
|
||||
stdout io.Writer
|
||||
|
||||
detections detectionsArgs
|
||||
detections detectionsArgs
|
||||
investigate investigateArgs
|
||||
}
|
||||
|
||||
// Initialize allows AccessGraphCommand to plug itself into the CLI parser.
|
||||
@@ -53,6 +54,7 @@ func (c *AccessGraphCommand) Initialize(app *kingpin.Application, cliFlags *tctl
|
||||
|
||||
// Initialize AG subcommands.
|
||||
c.initDetections(app)
|
||||
c.initInvestigate(app)
|
||||
}
|
||||
|
||||
// TryRun takes the CLI command as an argument and executes it.
|
||||
@@ -70,6 +72,8 @@ func (c *AccessGraphCommand) TryRun(ctx context.Context, cmd string, clientFunc
|
||||
commandFunc = c.DetectionsList
|
||||
case c.detections.get.cmd.FullCommand():
|
||||
commandFunc = c.DetectionsGet
|
||||
case c.investigate.cmd.FullCommand():
|
||||
commandFunc = c.Investigate
|
||||
default:
|
||||
return false, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,407 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2026 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package accessgraph
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/alecthomas/kingpin/v2"
|
||||
"github.com/gravitational/trace"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
"github.com/gravitational/teleport"
|
||||
accessgraph "github.com/gravitational/teleport/lib/accessgraph/apiclient"
|
||||
logmodels "github.com/gravitational/teleport/lib/accessgraph/apiclient/models/logs"
|
||||
)
|
||||
|
||||
// investigateArgs holds the parsed flag values for `tctl investigate`. The
|
||||
// include/exclude slices mirror the filter fields exposed by the Identity
|
||||
// Security investigate UI; see filterFields for the mapping to Lucene fields.
|
||||
type investigateArgs struct {
|
||||
cmd *kingpin.CmdClause
|
||||
|
||||
from time.Time
|
||||
to time.Time
|
||||
limit int
|
||||
order string
|
||||
format string
|
||||
|
||||
// Include filters
|
||||
includeIdentity []string
|
||||
includeUserKind []string
|
||||
includeEventType []string
|
||||
includeResource []string
|
||||
includeResourceKind []string
|
||||
includeIP []string
|
||||
includeStatus []string
|
||||
includeSource []string
|
||||
includeCountry []string
|
||||
includeCity []string
|
||||
includeRegion []string
|
||||
includeAWSAccountID []string
|
||||
includeAWSService []string
|
||||
includeGitHubOrg []string
|
||||
includeGitHubRepo []string
|
||||
includeOktaOrg []string
|
||||
includeTeleportCluster []string
|
||||
includeToken []string
|
||||
includeUserAgent []string
|
||||
|
||||
// Exclude filters
|
||||
excludeIdentity []string
|
||||
excludeUserKind []string
|
||||
excludeEventType []string
|
||||
excludeResource []string
|
||||
excludeResourceKind []string
|
||||
excludeIP []string
|
||||
excludeStatus []string
|
||||
excludeSource []string
|
||||
excludeCountry []string
|
||||
excludeCity []string
|
||||
excludeRegion []string
|
||||
excludeAWSAccountID []string
|
||||
excludeAWSService []string
|
||||
excludeGitHubOrg []string
|
||||
excludeGitHubRepo []string
|
||||
excludeOktaOrg []string
|
||||
excludeTeleportCluster []string
|
||||
excludeToken []string
|
||||
excludeUserAgent []string
|
||||
|
||||
// Raw query
|
||||
rawQuery string
|
||||
|
||||
// General flags
|
||||
printQuery bool
|
||||
showUnmatched bool
|
||||
facetsOnly bool
|
||||
}
|
||||
|
||||
// filterField describes one structured filter exposed by the CLI
|
||||
type filterField struct {
|
||||
flag string
|
||||
lucene string
|
||||
help string
|
||||
include *[]string
|
||||
exclude *[]string
|
||||
enumValues []string
|
||||
}
|
||||
|
||||
// filterFields enumerates every structured filter the CLI exposes.
|
||||
func (a *investigateArgs) filterFields() []filterField {
|
||||
return []filterField{
|
||||
{flag: "aws-account-id", lucene: "aws_account_id", help: "Filter by AWS account ID (repeatable).", include: &a.includeAWSAccountID, exclude: &a.excludeAWSAccountID},
|
||||
{flag: "aws-service", lucene: "aws_service", help: "Filter by AWS service name (repeatable).", include: &a.includeAWSService, exclude: &a.excludeAWSService},
|
||||
{flag: "city", lucene: "city", help: "Filter by city of origin (repeatable).", include: &a.includeCity, exclude: &a.excludeCity},
|
||||
{flag: "country", lucene: "country", help: "Filter by country of origin (repeatable).", include: &a.includeCountry, exclude: &a.excludeCountry},
|
||||
{flag: "event-type", lucene: "event_type", help: "Filter by event type, e.g. session.start (repeatable).", include: &a.includeEventType, exclude: &a.excludeEventType},
|
||||
{flag: "github-org", lucene: "github_organization", help: "Filter by GitHub organization (repeatable).", include: &a.includeGitHubOrg, exclude: &a.excludeGitHubOrg},
|
||||
{flag: "github-repo", lucene: "github_repo", help: "Filter by GitHub repository (repeatable).", include: &a.includeGitHubRepo, exclude: &a.excludeGitHubRepo},
|
||||
// We are using `--user` instead of `--identity` since the latter is already registered at the top level in tctl.
|
||||
{flag: "user", lucene: "identity_id", help: "Filter by user (email for users, ID for bots; repeatable).", include: &a.includeIdentity, exclude: &a.excludeIdentity},
|
||||
{flag: "user-kind", lucene: "identity_kind", help: "Filter by user kind, e.g. user, system (repeatable).", include: &a.includeUserKind, exclude: &a.excludeUserKind},
|
||||
{flag: "ip", lucene: "ip", help: "Filter by source IP address (repeatable).", include: &a.includeIP, exclude: &a.excludeIP},
|
||||
{flag: "okta-org", lucene: "okta_org", help: "Filter by Okta organization (repeatable).", include: &a.includeOktaOrg, exclude: &a.excludeOktaOrg},
|
||||
{flag: "region", lucene: "region", help: "Filter by region (e.g. us-east-1 or a US state code; repeatable).", include: &a.includeRegion, exclude: &a.excludeRegion},
|
||||
// Lucene names match the Athena column names directly so facet responses (which carry the Athena names) line up with our flags
|
||||
{flag: "resource", lucene: "target_resource", help: "Filter by target resource (repeatable).", include: &a.includeResource, exclude: &a.excludeResource},
|
||||
{flag: "resource-kind", lucene: "target_kind", help: "Filter by resource kind, e.g. ssh, kube, session_recording (repeatable).", include: &a.includeResourceKind, exclude: &a.excludeResourceKind},
|
||||
{flag: "source", lucene: "event_source", help: "Filter by event source (repeatable).", include: &a.includeSource, exclude: &a.excludeSource},
|
||||
{flag: "status", lucene: "status", help: "Filter by event status (repeatable).", include: &a.includeStatus, exclude: &a.excludeStatus, enumValues: []string{"success", "failure"}},
|
||||
{flag: "teleport-cluster", lucene: "teleport_cluster", help: "Filter by Teleport cluster name (repeatable).", include: &a.includeTeleportCluster, exclude: &a.excludeTeleportCluster},
|
||||
{flag: "token", lucene: "token", help: "Filter by token identifier (repeatable).", include: &a.includeToken, exclude: &a.excludeToken},
|
||||
{flag: "user-agent", lucene: "user_agent", help: "Filter by user agent string (repeatable). Not populated on every Teleport event — use deliberately.", include: &a.includeUserAgent, exclude: &a.excludeUserAgent},
|
||||
}
|
||||
}
|
||||
|
||||
// luceneToFlagMap returns a map from Lucene field name to flag name, so we can rename to match the CLI flags
|
||||
func (a *investigateArgs) luceneToFlagMap() map[string]string {
|
||||
fields := a.filterFields()
|
||||
names := make(map[string]string, len(fields))
|
||||
for _, f := range fields {
|
||||
names[f.lucene] = f.flag
|
||||
}
|
||||
return names
|
||||
}
|
||||
|
||||
// initInvestigate registers `tctl investigate` and all its flags.
|
||||
func (c *AccessGraphCommand) initInvestigate(app *kingpin.Application) {
|
||||
cmd := app.Command("investigate", "Search and explore Identity Security activity logs.")
|
||||
|
||||
cmd.Flag("from", fmt.Sprintf("Include activity at or after this time. (Examples: %s, %s, 24h, 7d; negative durations like -1h are future-relative. Default: 1d)", time.RFC3339, time.DateOnly)).
|
||||
Default("1d").
|
||||
SetValue(timeValue{target: &c.investigate.from})
|
||||
cmd.Flag("to", fmt.Sprintf("Include activity at or before this time. (Examples: %s, %s, 24h, 7d; negative durations like -1h are future-relative. Default: now)", time.RFC3339, time.DateOnly)).
|
||||
Default("now").
|
||||
SetValue(timeValue{target: &c.investigate.to})
|
||||
|
||||
cmd.Flag("limit", "Maximum number of events to return (0 for unlimited).").
|
||||
Default("100").
|
||||
IntVar(&c.investigate.limit)
|
||||
cmd.Flag("order", "Result order by timestamp. (Values: asc, desc)").
|
||||
Default(string(accessgraph.Desc)).
|
||||
EnumVar(&c.investigate.order, string(accessgraph.Asc), string(accessgraph.Desc))
|
||||
// TODO(ghassanachi): add text output option and set to default
|
||||
cmd.Flag("format", "Output format. (Values: json, yaml)").
|
||||
Default(teleport.YAML).
|
||||
EnumVar(&c.investigate.format, teleport.JSON, teleport.YAML)
|
||||
cmd.Flag("show-unmatched", "Include facet values that exist in the time window but did not match the current filter (the backend reports these with count=-1). Useful for discovering filters to broaden.").
|
||||
BoolVar(&c.investigate.showUnmatched)
|
||||
cmd.Flag("facets-only", "Skip fetching events; return only the facet summary. Useful for narrowing a query before pulling logs.").
|
||||
BoolVar(&c.investigate.facetsOnly)
|
||||
|
||||
for _, f := range c.investigate.filterFields() {
|
||||
include := cmd.Flag(f.flag, f.help)
|
||||
exclude := cmd.Flag("exclude-"+f.flag, "Exclude "+f.lucene+" values (repeatable).")
|
||||
if len(f.enumValues) > 0 {
|
||||
include.EnumsVar(f.include, f.enumValues...)
|
||||
exclude.EnumsVar(f.exclude, f.enumValues...)
|
||||
} else {
|
||||
include.StringsVar(f.include)
|
||||
exclude.StringsVar(f.exclude)
|
||||
}
|
||||
}
|
||||
|
||||
cmd.Flag("query", `Raw Lucene query. Mutually exclusive with structured filter flags. Example: --query 'identity_id:"alice@example.com" AND NOT status:"failure"'`).
|
||||
StringVar(&c.investigate.rawQuery)
|
||||
cmd.Flag("print-query", "Print the constructed query and exit without contacting the backend.").
|
||||
BoolVar(&c.investigate.printQuery)
|
||||
c.investigate.cmd = cmd
|
||||
}
|
||||
|
||||
// Investigate executes `tctl investigate`.
|
||||
func (c *AccessGraphCommand) Investigate(ctx context.Context, client *accessgraph.ClientWithResponses) error {
|
||||
args := &c.investigate
|
||||
|
||||
// Check raw-vs-structured conflict before --print-query: it shapes the query.
|
||||
if err := args.validateRawQueryExclusive(); err != nil {
|
||||
return trace.Wrap(err)
|
||||
}
|
||||
|
||||
query := args.buildQuery()
|
||||
|
||||
if args.printQuery {
|
||||
_, err := fmt.Fprintln(c.stdout, query)
|
||||
return trace.Wrap(err)
|
||||
}
|
||||
|
||||
if err := validateTimeWindow(args.from, args.to); err != nil {
|
||||
return trace.Wrap(err)
|
||||
}
|
||||
// Normalize to UTC before sending to the backend
|
||||
// non-UTC time shifts the stats window relative to the logs window
|
||||
fromUTC := args.from.UTC()
|
||||
toUTC := args.to.UTC()
|
||||
|
||||
order := accessgraph.ExecuteLogsQueryV1ParamsOrder(args.order)
|
||||
params := accessgraph.ExecuteLogsQueryV1Params{
|
||||
StartTime: &fromUTC,
|
||||
EndTime: &toUTC,
|
||||
Order: &order,
|
||||
}
|
||||
if query != "" {
|
||||
params.Query = &query
|
||||
}
|
||||
if args.limit > 0 {
|
||||
params.Limit = &args.limit
|
||||
}
|
||||
|
||||
// Facets and events have no inter-dependency, so fetch them in parallel.
|
||||
var (
|
||||
facets []logsFacet
|
||||
total int64
|
||||
events []logmodels.AccessgraphStorageV1alphaEvent
|
||||
truncated bool
|
||||
)
|
||||
g, gctx := errgroup.WithContext(ctx)
|
||||
g.Go(func() error {
|
||||
var err error
|
||||
facets, total, err = fetchLogsFacets(gctx, client, accessgraph.ExecuteLogsStatsQueryV1Params{
|
||||
StartTime: &fromUTC,
|
||||
EndTime: &toUTC,
|
||||
Query: params.Query,
|
||||
}, args.luceneToFlagMap())
|
||||
return trace.Wrap(err)
|
||||
})
|
||||
if !args.facetsOnly {
|
||||
g.Go(func() error {
|
||||
var err error
|
||||
events, truncated, err = fetchAllLogs(gctx, client, params, args.limit)
|
||||
return trace.Wrap(err)
|
||||
})
|
||||
}
|
||||
if err := g.Wait(); err != nil {
|
||||
return trace.Wrap(err)
|
||||
}
|
||||
|
||||
if !args.showUnmatched {
|
||||
facets = stripUnmatchedFacets(facets)
|
||||
}
|
||||
|
||||
// Always emit a non-nil slice so JSON renders "data": [] rather than null
|
||||
if events == nil {
|
||||
events = []logmodels.AccessgraphStorageV1alphaEvent{}
|
||||
}
|
||||
|
||||
output := investigateOutput{
|
||||
Total: total,
|
||||
Truncated: truncated,
|
||||
Facets: facets,
|
||||
Data: events,
|
||||
}
|
||||
|
||||
return writeOutput(c.stdout, output, args.format, func(w io.Writer) error {
|
||||
// TODO(ghassanachi): Replace with actual text output in child PR.
|
||||
return trace.NotImplemented("unreachable: text output not yet supported")
|
||||
})
|
||||
}
|
||||
|
||||
// stripUnmatchedFacets drops facet values with count=-1 (not in filtered response)
|
||||
func stripUnmatchedFacets(facets []logsFacet) []logsFacet {
|
||||
out := make([]logsFacet, 0, len(facets))
|
||||
for _, f := range facets {
|
||||
var kept []logsFacetValue
|
||||
for _, v := range f.Values {
|
||||
if v.Count < 0 {
|
||||
continue
|
||||
}
|
||||
kept = append(kept, v)
|
||||
}
|
||||
if len(kept) == 0 {
|
||||
continue
|
||||
}
|
||||
out = append(out, logsFacet{Name: f.Name, Values: kept})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// investigateOutput is the top-level shape returned by Investigate.
|
||||
type investigateOutput struct {
|
||||
// Total number of events matching the filter, derived from the event_type facet.
|
||||
Total int64 `json:"total" yaml:"total"`
|
||||
// Truncated is true when more events matched than were returned under --limit.
|
||||
Truncated bool `json:"truncated" yaml:"truncated"`
|
||||
// Facets is the list of filters that matched the query (unlimited), to be used for further filtering
|
||||
Facets []logsFacet `json:"facets" yaml:"facets"`
|
||||
// Data is the list of events matching the query, subject to --limit truncation
|
||||
Data []logmodels.AccessgraphStorageV1alphaEvent `json:"data" yaml:"data"`
|
||||
}
|
||||
|
||||
// logsFacet is one column of the stats response that we render as a facet.
|
||||
type logsFacet struct {
|
||||
Name string `json:"name" yaml:"name"`
|
||||
Values []logsFacetValue `json:"values" yaml:"values"`
|
||||
}
|
||||
|
||||
// logsFacetValue is one bucket in a facet.
|
||||
type logsFacetValue struct {
|
||||
Value string `json:"value" yaml:"value"`
|
||||
Count int64 `json:"count" yaml:"count"`
|
||||
}
|
||||
|
||||
// fetchLogsFacets calls ExecuteLogsStatsQueryV1 and transforms the response into a list of logsFacets
|
||||
func fetchLogsFacets(ctx context.Context, client *accessgraph.ClientWithResponses, params accessgraph.ExecuteLogsStatsQueryV1Params, luceneToFlagMap map[string]string) ([]logsFacet, int64, error) {
|
||||
resp, err := doRequest(client.ExecuteLogsStatsQueryV1WithResponse(ctx, ¶ms))
|
||||
if err != nil {
|
||||
return nil, 0, trace.Wrap(err)
|
||||
}
|
||||
if resp.JSON200 == nil {
|
||||
return nil, 0, trace.Errorf("received nil json response from Access Graph API")
|
||||
}
|
||||
var total int64
|
||||
byFlag := make(map[string][]logsFacetValue, len(resp.JSON200.Data))
|
||||
for _, column := range resp.JSON200.Data {
|
||||
if len(column.Values) == 0 {
|
||||
continue
|
||||
}
|
||||
// Event type is always set so we use it as the aggregate count for total results
|
||||
if column.ColumnName == "event_type" {
|
||||
for _, v := range column.Values {
|
||||
if v.Count > 0 {
|
||||
total += v.Count
|
||||
}
|
||||
}
|
||||
}
|
||||
flag, ok := luceneToFlagMap[column.ColumnName]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
values := make([]logsFacetValue, len(column.Values))
|
||||
for i, v := range column.Values {
|
||||
values[i] = logsFacetValue{Value: v.Value, Count: v.Count}
|
||||
}
|
||||
sort.SliceStable(values, func(i, j int) bool {
|
||||
return values[i].Count > values[j].Count
|
||||
})
|
||||
byFlag[flag] = values
|
||||
}
|
||||
flags := make([]string, 0, len(byFlag))
|
||||
for flag := range byFlag {
|
||||
flags = append(flags, flag)
|
||||
}
|
||||
sort.Strings(flags)
|
||||
facets := make([]logsFacet, 0, len(flags))
|
||||
for _, flag := range flags {
|
||||
facets = append(facets, logsFacet{Name: flag, Values: byFlag[flag]})
|
||||
}
|
||||
return facets, total, nil
|
||||
}
|
||||
|
||||
// buildQuery returns either the raw --query value or a query assembled from structured filters
|
||||
func (a *investigateArgs) buildQuery() string {
|
||||
if a.rawQuery != "" {
|
||||
return a.rawQuery
|
||||
}
|
||||
var parts []string
|
||||
for _, f := range a.filterFields() {
|
||||
if clause := dslClause(f.lucene, *f.include); clause != "" {
|
||||
parts = append(parts, clause)
|
||||
}
|
||||
if clause := dslClause(f.lucene, *f.exclude); clause != "" {
|
||||
parts = append(parts, "NOT "+clause)
|
||||
}
|
||||
}
|
||||
return strings.Join(parts, " AND ")
|
||||
}
|
||||
|
||||
// validateRawQueryExclusive rejects combinations of --query with any structured filter
|
||||
func (a *investigateArgs) validateRawQueryExclusive() error {
|
||||
if a.rawQuery == "" {
|
||||
return nil
|
||||
}
|
||||
var offenders []string
|
||||
for _, f := range a.filterFields() {
|
||||
if len(*f.include) > 0 {
|
||||
offenders = append(offenders, "--"+f.flag)
|
||||
}
|
||||
if len(*f.exclude) > 0 {
|
||||
offenders = append(offenders, "--exclude-"+f.flag)
|
||||
}
|
||||
}
|
||||
if len(offenders) == 0 {
|
||||
return nil
|
||||
}
|
||||
sort.Strings(offenders)
|
||||
return trace.BadParameter("--query is mutually exclusive with structured filter flags; remove: %s", strings.Join(offenders, ", "))
|
||||
}
|
||||
@@ -0,0 +1,573 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2026 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package accessgraph
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/alecthomas/kingpin/v2"
|
||||
"github.com/gravitational/trace"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gopkg.in/yaml.v3"
|
||||
|
||||
"github.com/gravitational/teleport"
|
||||
accessgraph "github.com/gravitational/teleport/lib/accessgraph/apiclient"
|
||||
)
|
||||
|
||||
// logsStatsQueryPath pins the AG path so a generated-client drift fails the test.
|
||||
const logsStatsQueryPath = accessGraphAPIPath + "graph/logs/v1/stats"
|
||||
|
||||
// TestFilterFieldsUnique pins that no two entries share a flag or lucene name.
|
||||
func TestFilterFieldsUnique(t *testing.T) {
|
||||
a := &investigateArgs{}
|
||||
flags := map[string]bool{}
|
||||
lucene := map[string]bool{}
|
||||
for _, f := range a.filterFields() {
|
||||
require.False(t, flags[f.flag], "duplicate flag %q", f.flag)
|
||||
require.False(t, lucene[f.lucene], "duplicate lucene %q", f.lucene)
|
||||
flags[f.flag] = true
|
||||
lucene[f.lucene] = true
|
||||
}
|
||||
}
|
||||
|
||||
func TestInvestigateBuildQuery(t *testing.T) {
|
||||
|
||||
t.Run("no filters → empty query", func(t *testing.T) {
|
||||
a := &investigateArgs{}
|
||||
require.Empty(t, a.buildQuery())
|
||||
})
|
||||
|
||||
t.Run("raw query short-circuits structured assembly", func(t *testing.T) {
|
||||
a := &investigateArgs{
|
||||
rawQuery: `identity_id:"alice"`,
|
||||
includeIdentity: []string{"bob"}, // ignored when rawQuery is set
|
||||
}
|
||||
require.Equal(t, `identity_id:"alice"`, a.buildQuery())
|
||||
})
|
||||
|
||||
t.Run("single include emits field:value", func(t *testing.T) {
|
||||
a := &investigateArgs{includeIdentity: []string{"alice@example.com"}}
|
||||
require.Equal(t, `identity_id:"alice@example.com"`, a.buildQuery())
|
||||
})
|
||||
|
||||
t.Run("multiple values on the same flag are OR'd", func(t *testing.T) {
|
||||
a := &investigateArgs{includeEventType: []string{"session.start", "session.end"}}
|
||||
require.Equal(t, `event_type:("session.start" OR "session.end")`, a.buildQuery())
|
||||
})
|
||||
|
||||
t.Run("exclude is wrapped in NOT", func(t *testing.T) {
|
||||
a := &investigateArgs{excludeStatus: []string{"success"}}
|
||||
require.Equal(t, `NOT status:"success"`, a.buildQuery())
|
||||
})
|
||||
|
||||
t.Run("include + exclude across fields → AND-joined", func(t *testing.T) {
|
||||
a := &investigateArgs{
|
||||
includeIdentity: []string{"alice"},
|
||||
excludeStatus: []string{"success"},
|
||||
}
|
||||
// filterFields() order (identity before status) pins the full string.
|
||||
require.Equal(t, `identity_id:"alice" AND NOT status:"success"`, a.buildQuery())
|
||||
})
|
||||
}
|
||||
|
||||
func TestInvestigateValidateRawQueryExclusive(t *testing.T) {
|
||||
|
||||
t.Run("no raw query → never errors", func(t *testing.T) {
|
||||
a := &investigateArgs{includeIdentity: []string{"alice"}}
|
||||
require.NoError(t, a.validateRawQueryExclusive())
|
||||
})
|
||||
|
||||
t.Run("raw query alone is allowed", func(t *testing.T) {
|
||||
a := &investigateArgs{rawQuery: `status:"failure"`}
|
||||
require.NoError(t, a.validateRawQueryExclusive())
|
||||
})
|
||||
|
||||
t.Run("raw query plus include → BadParameter listing offender", func(t *testing.T) {
|
||||
a := &investigateArgs{
|
||||
rawQuery: `status:"failure"`,
|
||||
includeIdentity: []string{"alice"},
|
||||
}
|
||||
err := a.validateRawQueryExclusive()
|
||||
require.True(t, trace.IsBadParameter(err), "want BadParameter, got %v", err)
|
||||
require.Contains(t, err.Error(), "--user")
|
||||
})
|
||||
|
||||
t.Run("raw query plus exclude → BadParameter listing offender", func(t *testing.T) {
|
||||
a := &investigateArgs{
|
||||
rawQuery: `status:"failure"`,
|
||||
excludeStatus: []string{"success"},
|
||||
}
|
||||
err := a.validateRawQueryExclusive()
|
||||
require.True(t, trace.IsBadParameter(err), "want BadParameter, got %v", err)
|
||||
require.Contains(t, err.Error(), "--exclude-status")
|
||||
})
|
||||
}
|
||||
|
||||
// TestStripUnmatchedFacets covers the count<0 sentinel: the backend uses
|
||||
// count == -1 for values present in the window but absent from the filter.
|
||||
func TestStripUnmatchedFacets(t *testing.T) {
|
||||
|
||||
t.Run("drops negative counts, keeps positive", func(t *testing.T) {
|
||||
in := []logsFacet{{Name: "user", Values: []logsFacetValue{
|
||||
{Value: "alice", Count: 5},
|
||||
{Value: "bob", Count: -1},
|
||||
{Value: "carol", Count: 2},
|
||||
}}}
|
||||
out := stripUnmatchedFacets(in)
|
||||
require.Len(t, out, 1)
|
||||
require.Equal(t, []logsFacetValue{
|
||||
{Value: "alice", Count: 5},
|
||||
{Value: "carol", Count: 2},
|
||||
}, out[0].Values)
|
||||
})
|
||||
|
||||
t.Run("facet with only unmatched values is dropped", func(t *testing.T) {
|
||||
in := []logsFacet{
|
||||
{Name: "user", Values: []logsFacetValue{{Value: "alice", Count: 5}}},
|
||||
{Name: "user-agent", Values: []logsFacetValue{{Value: "ua1", Count: -1}}},
|
||||
}
|
||||
out := stripUnmatchedFacets(in)
|
||||
require.Len(t, out, 1, "user-agent facet should have been dropped")
|
||||
require.Equal(t, "user", out[0].Name)
|
||||
})
|
||||
}
|
||||
|
||||
// statsResponse builds a stats endpoint body for table tests.
|
||||
func statsResponse(columns ...statsColumn) map[string]any {
|
||||
data := make([]map[string]any, 0, len(columns))
|
||||
for _, c := range columns {
|
||||
values := make([]map[string]any, 0, len(c.values))
|
||||
for _, v := range c.values {
|
||||
values = append(values, map[string]any{"value": v.value, "count": v.count})
|
||||
}
|
||||
data = append(data, map[string]any{
|
||||
"column_name": c.name,
|
||||
"values": values,
|
||||
})
|
||||
}
|
||||
return map[string]any{"data": data}
|
||||
}
|
||||
|
||||
type statsColumn struct {
|
||||
name string
|
||||
values []statsValue
|
||||
}
|
||||
|
||||
type statsValue struct {
|
||||
value string
|
||||
count int64
|
||||
}
|
||||
|
||||
func TestFetchLogsFacets(t *testing.T) {
|
||||
|
||||
a := &investigateArgs{}
|
||||
luceneToFlagMap := a.luceneToFlagMap()
|
||||
|
||||
t.Run("renames + sorts + drops non-filterable columns", func(t *testing.T) {
|
||||
body := statsResponse(
|
||||
// identity_id → "user", unsorted on the wire.
|
||||
statsColumn{name: "identity_id", values: []statsValue{
|
||||
{value: "bob", count: 1},
|
||||
{value: "alice", count: 7},
|
||||
{value: "carol", count: 3},
|
||||
}},
|
||||
// event_type drives the total and renders as "event-type".
|
||||
statsColumn{name: "event_type", values: []statsValue{
|
||||
{value: "session.start", count: 4},
|
||||
{value: "session.end", count: 5},
|
||||
}},
|
||||
// Non-filterable, must be dropped.
|
||||
statsColumn{name: "row_count", values: []statsValue{
|
||||
{value: "irrelevant", count: 99},
|
||||
}},
|
||||
// Empty values, must be dropped.
|
||||
statsColumn{name: "status", values: nil},
|
||||
)
|
||||
ag := newAccessGraphTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
require.Equal(t, logsStatsQueryPath, r.URL.Path)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(body)
|
||||
}))
|
||||
|
||||
facets, total, err := fetchLogsFacets(context.Background(), ag,
|
||||
accessgraph.ExecuteLogsStatsQueryV1Params{}, luceneToFlagMap)
|
||||
require.NoError(t, err)
|
||||
require.EqualValues(t, 9, total, "sum of event_type counts (4 + 5)")
|
||||
|
||||
names := make([]string, len(facets))
|
||||
for i, f := range facets {
|
||||
names[i] = f.Name
|
||||
}
|
||||
require.Equal(t, []string{"event-type", "user"}, names)
|
||||
|
||||
userFacet := facets[1]
|
||||
require.Equal(t, []logsFacetValue{
|
||||
{Value: "alice", Count: 7},
|
||||
{Value: "carol", Count: 3},
|
||||
{Value: "bob", Count: 1},
|
||||
}, userFacet.Values, "sorted by count desc")
|
||||
})
|
||||
|
||||
t.Run("negative counts in event_type don't inflate total", func(t *testing.T) {
|
||||
body := statsResponse(statsColumn{name: "event_type", values: []statsValue{
|
||||
{value: "session.start", count: 4},
|
||||
{value: "session.end", count: -1},
|
||||
}})
|
||||
ag := newAccessGraphTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(body)
|
||||
}))
|
||||
_, total, err := fetchLogsFacets(context.Background(), ag,
|
||||
accessgraph.ExecuteLogsStatsQueryV1Params{}, luceneToFlagMap)
|
||||
require.NoError(t, err)
|
||||
require.EqualValues(t, 4, total)
|
||||
})
|
||||
}
|
||||
|
||||
var (
|
||||
investigateFixtureFrom = time.Date(2026, 4, 1, 12, 0, 0, 0, time.UTC)
|
||||
investigateFixtureTo = time.Date(2026, 4, 2, 12, 0, 0, 0, time.UTC)
|
||||
)
|
||||
|
||||
// newInvestigateCommand returns a command wired to a captured buffer. Tests
|
||||
// call Investigate directly, bypassing TryRun's credential loading.
|
||||
func newInvestigateCommand(t *testing.T, format string) (*AccessGraphCommand, *bytes.Buffer) {
|
||||
t.Helper()
|
||||
var buf bytes.Buffer
|
||||
c := &AccessGraphCommand{
|
||||
stdout: &buf,
|
||||
investigate: investigateArgs{
|
||||
format: format,
|
||||
from: investigateFixtureFrom,
|
||||
to: investigateFixtureTo,
|
||||
order: string(accessgraph.Desc),
|
||||
limit: 100,
|
||||
},
|
||||
}
|
||||
return c, &buf
|
||||
}
|
||||
|
||||
// investigateHandler serves the AG stats and logs endpoints; nil logsPages
|
||||
// asserts the logs route is never hit.
|
||||
type investigateHandler struct {
|
||||
stats map[string]any
|
||||
statsCalls atomic.Int64
|
||||
statsQuery string
|
||||
statsStart string
|
||||
statsEnd string
|
||||
logsPages []fetchAllLogsPage
|
||||
logsCalls atomic.Int64
|
||||
logsQuery string
|
||||
logsStart string
|
||||
logsEnd string
|
||||
logsOrder string
|
||||
logsLatitude string
|
||||
logsLongitude string
|
||||
logsRadius string
|
||||
}
|
||||
|
||||
func (h *investigateHandler) serve(t *testing.T) http.Handler {
|
||||
t.Helper()
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch r.URL.Path {
|
||||
case logsStatsQueryPath:
|
||||
h.statsCalls.Add(1)
|
||||
h.statsQuery = r.URL.Query().Get("query")
|
||||
h.statsStart = r.URL.Query().Get("start_time")
|
||||
h.statsEnd = r.URL.Query().Get("end_time")
|
||||
_ = json.NewEncoder(w).Encode(h.stats)
|
||||
case logsQueryPath:
|
||||
idx := int(h.logsCalls.Add(1) - 1)
|
||||
if idx == 0 {
|
||||
h.logsQuery = r.URL.Query().Get("query")
|
||||
h.logsStart = r.URL.Query().Get("start_time")
|
||||
h.logsEnd = r.URL.Query().Get("end_time")
|
||||
h.logsOrder = r.URL.Query().Get("order")
|
||||
h.logsLatitude = r.URL.Query().Get("latitude")
|
||||
h.logsLongitude = r.URL.Query().Get("longitude")
|
||||
h.logsRadius = r.URL.Query().Get("radius")
|
||||
}
|
||||
require.Less(t, idx, len(h.logsPages), "more pages requested than configured")
|
||||
page := h.logsPages[idx]
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
"data": page.data,
|
||||
"next_cursor": page.nextCursor,
|
||||
})
|
||||
default:
|
||||
t.Errorf("unexpected request to %s", r.URL.Path)
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestInvestigate(t *testing.T) {
|
||||
|
||||
t.Run("--print-query prints the assembled query and skips the backend", func(t *testing.T) {
|
||||
ag := newAccessGraphTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
t.Errorf("server reached despite --print-query: %s", r.URL.Path)
|
||||
}))
|
||||
c, buf := newInvestigateCommand(t, teleport.Text)
|
||||
c.investigate.includeIdentity = []string{"alice"}
|
||||
c.investigate.excludeStatus = []string{"success"}
|
||||
c.investigate.printQuery = true
|
||||
|
||||
require.NoError(t, c.Investigate(context.Background(), ag))
|
||||
require.Equal(t, `identity_id:"alice" AND NOT status:"success"`,
|
||||
strings.TrimSpace(buf.String()))
|
||||
})
|
||||
|
||||
t.Run("invalid time window rejected before backend is hit", func(t *testing.T) {
|
||||
ag := newAccessGraphTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
t.Errorf("server reached despite invalid window: %s", r.URL.Path)
|
||||
}))
|
||||
c, _ := newInvestigateCommand(t, teleport.JSON)
|
||||
c.investigate.from = investigateFixtureTo
|
||||
c.investigate.to = investigateFixtureFrom
|
||||
err := c.Investigate(context.Background(), ag)
|
||||
require.True(t, trace.IsBadParameter(err), "want BadParameter, got %v", err)
|
||||
})
|
||||
|
||||
t.Run("full flow sends UTC window and structured query to both endpoints", func(t *testing.T) {
|
||||
h := &investigateHandler{
|
||||
stats: statsResponse(
|
||||
statsColumn{name: "event_type", values: []statsValue{{value: "session.start", count: 3}}},
|
||||
statsColumn{name: "identity_id", values: []statsValue{{value: "alice", count: 3}}},
|
||||
),
|
||||
logsPages: []fetchAllLogsPage{{data: []map[string]any{
|
||||
{"uuid": "11111111-1111-1111-1111-111111111111"},
|
||||
}}},
|
||||
}
|
||||
ag := newAccessGraphTestClient(t, h.serve(t))
|
||||
|
||||
c, _ := newInvestigateCommand(t, teleport.JSON)
|
||||
// Pin the from/to to a non-UTC zone so we can prove .UTC() conversion happens.
|
||||
zone := time.FixedZone("UTC-7", -7*3600)
|
||||
c.investigate.from = investigateFixtureFrom.In(zone)
|
||||
c.investigate.to = investigateFixtureTo.In(zone)
|
||||
c.investigate.includeIdentity = []string{"alice"}
|
||||
|
||||
require.NoError(t, c.Investigate(context.Background(), ag))
|
||||
|
||||
require.EqualValues(t, 1, h.statsCalls.Load())
|
||||
require.EqualValues(t, 1, h.logsCalls.Load())
|
||||
require.Equal(t, `identity_id:"alice"`, h.statsQuery)
|
||||
require.Equal(t, `identity_id:"alice"`, h.logsQuery)
|
||||
|
||||
// Stats reinterprets the literal as UTC, so a non-UTC time would
|
||||
// shift the stats window relative to the logs window.
|
||||
gotStart, err := time.Parse(time.RFC3339Nano, h.statsStart)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, time.UTC, gotStart.Location())
|
||||
require.True(t, gotStart.Equal(investigateFixtureFrom))
|
||||
|
||||
gotLogsStart, err := time.Parse(time.RFC3339Nano, h.logsStart)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, time.UTC, gotLogsStart.Location())
|
||||
|
||||
require.Equal(t, "desc", h.logsOrder)
|
||||
require.Empty(t, h.logsLatitude)
|
||||
})
|
||||
|
||||
t.Run("--facets-only skips the logs endpoint and emits empty events array", func(t *testing.T) {
|
||||
h := &investigateHandler{
|
||||
stats: statsResponse(
|
||||
statsColumn{name: "event_type", values: []statsValue{{value: "session.start", count: 2}}},
|
||||
),
|
||||
}
|
||||
ag := newAccessGraphTestClient(t, h.serve(t))
|
||||
|
||||
c, buf := newInvestigateCommand(t, teleport.JSON)
|
||||
c.investigate.facetsOnly = true
|
||||
|
||||
require.NoError(t, c.Investigate(context.Background(), ag))
|
||||
require.EqualValues(t, 1, h.statsCalls.Load())
|
||||
require.EqualValues(t, 0, h.logsCalls.Load())
|
||||
|
||||
var out struct {
|
||||
Total int64 `json:"total"`
|
||||
Data []json.RawMessage `json:"data"`
|
||||
}
|
||||
require.NoError(t, json.Unmarshal(buf.Bytes(), &out))
|
||||
require.EqualValues(t, 2, out.Total)
|
||||
require.Empty(t, out.Data)
|
||||
|
||||
// "data" must encode as [], never null, so jq consumers can `.data[]`.
|
||||
compact := strings.Join(strings.Fields(buf.String()), "")
|
||||
require.Contains(t, compact, `"data":[]`)
|
||||
require.NotContains(t, compact, `"data":null`)
|
||||
})
|
||||
|
||||
t.Run("--show-unmatched preserves count=-1 values", func(t *testing.T) {
|
||||
h := &investigateHandler{
|
||||
stats: statsResponse(
|
||||
statsColumn{name: "event_type", values: []statsValue{{value: "session.start", count: 4}}},
|
||||
statsColumn{name: "identity_id", values: []statsValue{
|
||||
{value: "alice", count: 4},
|
||||
{value: "bob", count: -1},
|
||||
}},
|
||||
),
|
||||
logsPages: []fetchAllLogsPage{{data: nil}},
|
||||
}
|
||||
ag := newAccessGraphTestClient(t, h.serve(t))
|
||||
|
||||
c, buf := newInvestigateCommand(t, teleport.JSON)
|
||||
c.investigate.showUnmatched = true
|
||||
require.NoError(t, c.Investigate(context.Background(), ag))
|
||||
|
||||
var out investigateOutput
|
||||
require.NoError(t, json.Unmarshal(buf.Bytes(), &out))
|
||||
var userFacet *logsFacet
|
||||
for i := range out.Facets {
|
||||
if out.Facets[i].Name == "user" {
|
||||
userFacet = &out.Facets[i]
|
||||
}
|
||||
}
|
||||
require.NotNil(t, userFacet)
|
||||
require.Len(t, userFacet.Values, 2)
|
||||
// Sorted by count desc, so unmatched (-1) trails the positive count.
|
||||
require.EqualValues(t, 4, userFacet.Values[0].Count)
|
||||
require.EqualValues(t, -1, userFacet.Values[1].Count)
|
||||
})
|
||||
|
||||
t.Run("yaml format round-trips total + facets + data", func(t *testing.T) {
|
||||
h := &investigateHandler{
|
||||
stats: statsResponse(
|
||||
statsColumn{name: "event_type", values: []statsValue{{value: "session.start", count: 1}}},
|
||||
),
|
||||
logsPages: []fetchAllLogsPage{{data: []map[string]any{
|
||||
{"uuid": "11111111-1111-1111-1111-111111111111"},
|
||||
}}},
|
||||
}
|
||||
ag := newAccessGraphTestClient(t, h.serve(t))
|
||||
|
||||
c, buf := newInvestigateCommand(t, teleport.YAML)
|
||||
require.NoError(t, c.Investigate(context.Background(), ag))
|
||||
|
||||
var out investigateOutput
|
||||
require.NoError(t, yaml.Unmarshal(buf.Bytes(), &out))
|
||||
require.EqualValues(t, 1, out.Total)
|
||||
require.Len(t, out.Data, 1)
|
||||
})
|
||||
|
||||
t.Run("--limit caps events and surfaces truncated", func(t *testing.T) {
|
||||
h := &investigateHandler{
|
||||
stats: statsResponse(
|
||||
statsColumn{name: "event_type", values: []statsValue{{value: "session.start", count: 2}}},
|
||||
),
|
||||
// One page with more events than --limit, so fetchAllLogs trims and flags truncated.
|
||||
logsPages: []fetchAllLogsPage{{data: []map[string]any{
|
||||
{"uuid": "11111111-1111-1111-1111-111111111111"},
|
||||
{"uuid": "22222222-2222-2222-2222-222222222222"},
|
||||
}}},
|
||||
}
|
||||
ag := newAccessGraphTestClient(t, h.serve(t))
|
||||
|
||||
c, buf := newInvestigateCommand(t, teleport.JSON)
|
||||
c.investigate.limit = 1
|
||||
|
||||
require.NoError(t, c.Investigate(context.Background(), ag))
|
||||
|
||||
var out investigateOutput
|
||||
require.NoError(t, json.Unmarshal(buf.Bytes(), &out))
|
||||
require.True(t, out.Truncated)
|
||||
require.Len(t, out.Data, 1)
|
||||
})
|
||||
|
||||
t.Run("stats HTTP 500 surfaces as apiResponseError", func(t *testing.T) {
|
||||
ag := newAccessGraphTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case logsStatsQueryPath:
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
_, _ = w.Write([]byte(`{"message":"stats kaput"}`))
|
||||
case logsQueryPath:
|
||||
// Logs runs in parallel with stats; race-tolerant empty response.
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"data":[]}`))
|
||||
}
|
||||
}))
|
||||
c, _ := newInvestigateCommand(t, teleport.JSON)
|
||||
err := c.Investigate(context.Background(), ag)
|
||||
var agErr *apiResponseError
|
||||
require.ErrorAs(t, err, &agErr)
|
||||
require.Equal(t, http.StatusInternalServerError, agErr.StatusCode)
|
||||
require.Equal(t, "stats kaput", agErr.Message)
|
||||
})
|
||||
}
|
||||
|
||||
// TestInitInvestigateFlags exercises kingpin wiring without going through TryRun.
|
||||
func TestInitInvestigateFlags(t *testing.T) {
|
||||
|
||||
// parse builds a fresh app per subtest so parser state is isolated.
|
||||
parse := func(t *testing.T, argv ...string) (investigateArgs, error) {
|
||||
t.Helper()
|
||||
app := kingpin.New("tctl-test", "")
|
||||
c := &AccessGraphCommand{}
|
||||
c.initInvestigate(app)
|
||||
_, err := app.Parse(argv)
|
||||
return c.investigate, err
|
||||
}
|
||||
|
||||
t.Run("defaults", func(t *testing.T) {
|
||||
before := time.Now()
|
||||
got, err := parse(t, "investigate")
|
||||
after := time.Now()
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Equal(t, teleport.YAML, got.format)
|
||||
require.Equal(t, string(accessgraph.Desc), got.order)
|
||||
require.Equal(t, 100, got.limit)
|
||||
require.False(t, got.facetsOnly)
|
||||
|
||||
// --from defaults to "1d"; widened slack absorbs CI clock drift.
|
||||
require.WithinRange(t, got.from,
|
||||
before.Add(-24*time.Hour-5*time.Second),
|
||||
after.Add(-24*time.Hour+5*time.Second))
|
||||
require.WithinRange(t, got.to, before.Add(-5*time.Second), after.Add(5*time.Second))
|
||||
})
|
||||
|
||||
t.Run("structured filters populate include/exclude slices", func(t *testing.T) {
|
||||
got, err := parse(t,
|
||||
"investigate",
|
||||
"--user", "alice",
|
||||
"--user", "bob",
|
||||
"--exclude-status", "success",
|
||||
"--user-agent", "Mozilla/5.0",
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, []string{"alice", "bob"}, got.includeIdentity)
|
||||
require.Equal(t, []string{"success"}, got.excludeStatus)
|
||||
require.Equal(t, []string{"Mozilla/5.0"}, got.includeUserAgent)
|
||||
})
|
||||
|
||||
t.Run("boolean flags flip to true", func(t *testing.T) {
|
||||
got, err := parse(t, "investigate", "--show-unmatched", "--facets-only")
|
||||
require.NoError(t, err)
|
||||
require.True(t, got.showUnmatched)
|
||||
require.True(t, got.facetsOnly)
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user