This is an automated email from the ASF dual-hosted git repository. github-actions[bot] pushed a commit to tag nightly in repository https://gitbox.apache.org/repos/asf/casbin-gateway.git
commit 04b0aab6a6ef8e66e89c4ff189a1b17a1dc222c4 Author: Paperlz <[email protected]> AuthorDate: Sat Aug 22 15:44:27 2026 +0800 feat: support Hermes Agent monitoring (#221) --- agentpatch/hermes_agent.go | 329 ++++++++++++++++++++++++++ agentpatch/hermes_config.go | 515 ++++++++++++++++++++++++++++++++++++++++ agentpatch/hermes_observer.py | 530 ++++++++++++++++++++++++++++++++++++++++++ agentpatch/hermes_plugin.yaml | 18 ++ go.mod | 1 + 5 files changed, 1393 insertions(+) diff --git a/agentpatch/hermes_agent.go b/agentpatch/hermes_agent.go new file mode 100644 index 0000000..4d1b817 --- /dev/null +++ b/agentpatch/hermes_agent.go @@ -0,0 +1,329 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed 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 agentpatch + +import ( + "bytes" + _ "embed" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "runtime" + "strings" + + "github.com/apache/casbin-gateway/agenthome" + "github.com/apache/casbin-gateway/agentmonitor" + "github.com/apache/casbin-gateway/internal/hermes" +) + +const ( + hermesPluginName = gatewayEntryName + hermesPluginVersion = "1.1.0" + hermesConfigSchema = 1 + minHermesPluginVersion = "0.19.0" +) + +//go:embed hermes_observer.py +var hermesObserver []byte + +//go:embed hermes_plugin.yaml +var hermesPluginManifest []byte + +type hermesPatcher struct{} + +type hermesLayout struct { + home string + configPath string + pluginDir string +} + +type hermesObserverConfig struct { + SchemaVersion int `json:"schemaVersion"` + Owner string `json:"owner"` + PluginVersion string `json:"pluginVersion"` + RecordsURL string `json:"recordsUrl"` + AgentPath string `json:"agentPath"` + User string `json:"user"` + IngestToken string `json:"ingestToken"` + IngestTokenHeader string `json:"ingestTokenHeader"` +} + +func init() { + register(hermesPatcher{}) +} + +func (hermesPatcher) AgentId() string { return hermes.AgentID } + +func (hermesPatcher) Supported() bool { return true } + +func (p hermesPatcher) Patch(target Target) error { + layout, err := p.validateTarget(target) + if err != nil { + return err + } + applied := IsApplied(target) + if info, statErr := os.Stat(layout.pluginDir); statErr == nil { + if !info.IsDir() { + return fmt.Errorf("%s exists and is not a directory", layout.pluginDir) + } + if !applied { + return fmt.Errorf("%s already exists and is not owned by this Gateway patch state", layout.pluginDir) + } + if owned, ownershipErr := hermesPluginOwned(layout.pluginDir); ownershipErr != nil { + return ownershipErr + } else if !owned { + return fmt.Errorf("%s no longer contains a Gateway-owned observer; refusing to refresh it", layout.pluginDir) + } + } else if !os.IsNotExist(statErr) { + return statErr + } + + token, err := IssueIngestToken(target) + if err != nil { + return err + } + observerConfig, err := renderHermesObserverConfig(target, token) + if err != nil { + return errors.Join(err, RevokeIngestToken(target)) + } + pluginFiles := []struct { + name string + data []byte + mode os.FileMode + }{ + {"plugin.yaml", hermesPluginManifest, 0o644}, + {"__init__.py", hermesObserver, 0o644}, + {"gateway.json", observerConfig, 0o600}, + } + + if err := Apply(target, func(changes *ChangeSet) error { + if err := changes.MkdirAll(layout.pluginDir); err != nil { + return err + } + for _, file := range pluginFiles { + path := filepath.Join(layout.pluginDir, file.name) + if err := changes.WriteFile(path, file.data, file.mode); err != nil { + return err + } + } + return nil + }); err != nil { + return errors.Join(err, RevokeIngestToken(target)) + } + + // config.yaml stays out of the backup journal on purpose. It is the user's + // own file and keeps changing for as long as the patch is on, so Unpatch + // deletes the one list entry Gateway added instead of restoring a whole + // file snapshot taken at Patch time. Only the installed plugin, which is + // entirely Gateway's, is journalled - and that is what undoes this if + // enabling the plugin fails. + if err := writeHermesPluginMembership(layout.configPath, true); err != nil { + return errors.Join(err, Revert(target), RevokeIngestToken(target)) + } + return nil +} + +func (p hermesPatcher) Unpatch(target Target) error { + // Revert works from the saved paths and does not need the current launcher + // to still be valid, so the layout is needed only to reach config.yaml. A + // host whose home directory no longer resolves still gets its plugin + // removed; the entry left in config.yaml then names a plugin that is gone, + // which Hermes ignores. + var membershipErr error + if layout, err := p.layoutOf(target); err == nil { + membershipErr = writeHermesPluginMembership(layout.configPath, false) + } + return errors.Join(membershipErr, Revert(target), RevokeIngestToken(target)) +} + +// PatchNotice is the copy the agents table shows before and after the button, +// so the Web UI does not need a branch of its own for how Hermes loads plugins. +func (hermesPatcher) PatchNotice(patched bool) (string, string) { + if patched { + return "Removes the behaviour observer from the default Hermes profile and drops its entry from plugins.enabled. Nothing else in config.yaml changes.", + "Restart running Hermes processes to unload it." + } + return "Installs a behaviour observer with redacted tool inputs in the default Hermes profile and adds it to plugins.enabled. Named profiles are left alone.", + "Restart running Hermes processes to load it." +} + +func (p hermesPatcher) Status(target Target) (Status, error) { + layout, err := p.layoutOf(target) + if err != nil { + return Status{}, err + } + config, owned, err := readHermesObserverConfig(layout.pluginDir) + if err != nil { + return Status{}, err + } + if !owned { + if _, statErr := os.Stat(layout.pluginDir); statErr == nil { + return Status{Detail: "plugin directory exists but is not owned by Gateway"}, nil + } + return Status{Detail: "not patched"}, nil + } + + issuedAgent, tokenValid := ValidateIngestToken(config.IngestToken) + if !tokenValid || !strings.EqualFold(issuedAgent, target.AgentId) { + return Status{Detail: "observer files are stale or incomplete; Patch again to refresh them"}, nil + } + expectedConfig, err := renderHermesObserverConfig(target, config.IngestToken) + if err != nil { + return Status{}, err + } + expectedFiles := map[string][]byte{ + "plugin.yaml": hermesPluginManifest, "__init__.py": hermesObserver, + "gateway.json": expectedConfig, + } + for name, expected := range expectedFiles { + actual, readErr := os.ReadFile(filepath.Join(layout.pluginDir, name)) + if readErr != nil || !bytes.Equal(actual, expected) { + return Status{Detail: "observer files are stale or incomplete; Patch again to refresh them"}, nil + } + } + enabled, disabled, err := hermesPluginMembership(layout.configPath) + if err != nil { + return Status{}, err + } + if !enabled || disabled { + return Status{Detail: "observer is installed but not enabled in config.yaml; Patch again to repair it"}, nil + } + if !IsApplied(target) { + return Status{ + Patched: true, + Detail: "observer is enabled outside this Gateway patch state; an automatic Unpatch is unsafe", + }, nil + } + return Status{ + Patched: true, + Detail: fmt.Sprintf( + "behaviour observer installed in %s; start a new Hermes process or restart a running one to load it", + layout.home, + ), + }, nil +} + +func (p hermesPatcher) validateTarget(target Target) (hermesLayout, error) { + layout, err := p.layoutOf(target) + if err != nil { + return hermesLayout{}, err + } + roots := []string{filepath.Join(layout.home, hermes.ProjectDir)} + if runtime.GOOS == "linux" && filepath.Clean(target.Path) == "/usr/local/bin/hermes" { + roots = append(roots, "/usr/local/lib/hermes-agent", "/root/.hermes/hermes-agent") + } + project, err := hermes.InspectLauncher(target.Path, roots...) + if err != nil { + return hermesLayout{}, err + } + // The version gate is the compatibility check. InspectLauncher has already + // confirmed the launcher belongs to a real hermes-agent project, and every + // release from here on exposes the observer hooks the plugin registers; + // a hook Hermes does not know simply never fires, which the fail-open + // observer is built for. + if !hermes.VersionAtLeast(project.Version, minHermesPluginVersion) { + return hermesLayout{}, fmt.Errorf( + "Hermes Agent %s is incompatible; upgrade to %s or later", + project.Version, minHermesPluginVersion, + ) + } + return layout, nil +} + +func (hermesPatcher) layoutOf(target Target) (hermesLayout, error) { + home, err := agenthome.Resolve(target.Owner) + if err != nil { + return hermesLayout{}, err + } + hermesHome := filepath.Join(home, ".hermes") + if runtime.GOOS == "windows" { + hermesHome = filepath.Join(home, "AppData", "Local", "hermes") + if isProcessHome(home) { + if local := os.Getenv("LOCALAPPDATA"); local != "" { + hermesHome = filepath.Join(local, "hermes") + } + } + } + if configured := os.Getenv("HERMES_HOME"); configured != "" && isProcessHome(home) { + hermesHome = configured + } + return hermesLayout{ + home: hermesHome, + configPath: filepath.Join(hermesHome, "config.yaml"), + pluginDir: filepath.Join(hermesHome, "plugins", hermesPluginName), + }, nil +} + +func renderHermesObserverConfig(target Target, token string) ([]byte, error) { + url, err := recordsURL() + if err != nil { + return nil, err + } + data, err := json.MarshalIndent(hermesObserverConfig{ + SchemaVersion: hermesConfigSchema, + Owner: "casbin-gateway", + PluginVersion: hermesPluginVersion, + RecordsURL: url, + AgentPath: target.Path, + User: target.Owner, + IngestToken: token, + IngestTokenHeader: agentmonitor.IngestTokenHeader, + }, "", " ") + if err != nil { + return nil, err + } + return append(data, '\n'), nil +} + +func hermesPluginOwned(pluginDir string) (bool, error) { + _, owned, err := readHermesObserverConfig(pluginDir) + return owned, err +} + +func readHermesObserverConfig(pluginDir string) (hermesObserverConfig, bool, error) { + data, err := os.ReadFile(filepath.Join(pluginDir, "gateway.json")) + if os.IsNotExist(err) { + return hermesObserverConfig{}, false, nil + } + if err != nil { + return hermesObserverConfig{}, false, err + } + var config hermesObserverConfig + if err := json.Unmarshal(data, &config); err != nil { + return hermesObserverConfig{}, false, nil + } + return config, config.Owner == "casbin-gateway" && + config.SchemaVersion == hermesConfigSchema, nil +} + +func isProcessHome(home string) bool { + current, err := os.UserHomeDir() + if err != nil { + return false + } + left, right := filepath.Clean(home), filepath.Clean(current) + if resolved, err := filepath.EvalSymlinks(left); err == nil { + left = resolved + } + if resolved, err := filepath.EvalSymlinks(right); err == nil { + right = resolved + } + if runtime.GOOS == "windows" { + return strings.EqualFold(left, right) + } + return left == right +} diff --git a/agentpatch/hermes_config.go b/agentpatch/hermes_config.go new file mode 100644 index 0000000..d1d6a00 --- /dev/null +++ b/agentpatch/hermes_config.go @@ -0,0 +1,515 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed 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 agentpatch + +// Hermes' config.yaml belongs to the user, not to Gateway: a normal install is +// well over a thousand lines of section banners, trailing notes and blank-line +// separated blocks. Round-tripping that through a YAML marshaller to flip one +// list entry rewrites every line of the file, so enabling the observer inserts +// a single list item into the text instead. Everything outside that one line +// stays byte for byte identical, which is also what lets Unpatch delete just +// that line rather than restore a whole-file snapshot taken at Patch time. +// +// Reading stays with the real YAML parser: hermesPluginMembership is what +// Status trusts, and writeHermesPluginMembership re-parses its own output +// before committing it, so a document shape the line editor gets wrong is +// caught here instead of by Hermes at startup. + +import ( + "bytes" + "errors" + "fmt" + "os" + "strings" + + "gopkg.in/yaml.v3" +) + +// writeHermesPluginMembership adds the observer to plugins.enabled, or removes +// it, editing configPath in place. An existing config keeps its current mode. +func writeHermesPluginMembership(configPath string, enabled bool) error { + data, err := os.ReadFile(configPath) + missing := os.IsNotExist(err) + if err != nil && !missing { + return err + } + if missing && !enabled { + return nil + } + + updated, changed, err := setHermesPluginMembership(data, enabled) + if err != nil { + return fmt.Errorf("update %s: %w", configPath, err) + } + if !changed { + return nil + } + if err := verifyHermesPluginMembership(updated, enabled); err != nil { + return fmt.Errorf("update %s: %w", configPath, err) + } + + if err := os.WriteFile(configPath, updated, 0o600); err != nil { + return err + } + return nil +} + +// verifyHermesPluginMembership re-reads an edited document with the YAML parser +// and checks the observer ended up where the edit intended. It is the guard +// that keeps a line edit from writing a document Hermes would read differently. +func verifyHermesPluginMembership(data []byte, enabled bool) error { + listed, blocked, err := hermesPluginMembershipOf(data) + if err != nil { + return err + } + if listed != enabled || (enabled && blocked) { + return errors.New("the edited plugins list does not read back as expected") + } + return nil +} + +// hermesPluginEdit is one list membership the observer needs in config.yaml. +type hermesPluginEdit struct { + key string + want bool +} + +// setHermesPluginMembership returns config.yaml with the observer added to or +// removed from plugins.enabled. Enabling also drops it from plugins.disabled, +// which Hermes treats as a deny list that wins over plugins.enabled. Disabling +// leaves plugins.disabled alone: an entry there is the user's, never ours. +func setHermesPluginMembership(data []byte, enabled bool) ([]byte, bool, error) { + edits := []hermesPluginEdit{{key: "enabled", want: enabled}} + if enabled { + edits = append(edits, hermesPluginEdit{key: "disabled", want: false}) + } + + current, changed := data, false + for _, edit := range edits { + next, stepChanged, err := setHermesPluginListMember(current, edit.key, edit.want) + if err != nil { + return nil, false, err + } + if stepChanged { + current, changed = next, true + } + } + return current, changed, nil +} + +// setHermesPluginListMember adds or removes the observer in plugins.<key>. The +// document is re-read for each edit so one edit's line numbers never depend on +// another's. +func setHermesPluginListMember(data []byte, key string, want bool) ([]byte, bool, error) { + document := newYAMLLines(data) + + plugins := document.findMappingKey("plugins", 0, 0, len(document.lines)) + if plugins < 0 { + if !want { + return data, false, nil + } + document.appendBlock("plugins:", " "+key+":", " - "+hermesPluginName) + return document.bytes(), true, nil + } + if _, value, _ := yamlMappingKey(document.lines[plugins], 0); value != "" { + return nil, false, fmt.Errorf("plugins must be a block mapping, found %q", value) + } + + pluginsEnd := document.blockEnd(0, plugins+1) + childIndent := document.childIndent(plugins, 0, pluginsEnd) + keyLine := document.findMappingKey(key, childIndent, plugins+1, pluginsEnd) + if keyLine < 0 { + if !want { + return data, false, nil + } + indent := strings.Repeat(" ", childIndent) + document.insert(document.lastContent(plugins+1, pluginsEnd)+1, + indent+key+":", indent+" - "+hermesPluginName) + return document.bytes(), true, nil + } + if _, value, _ := yamlMappingKey(document.lines[keyLine], childIndent); value != "" { + return setHermesPluginFlowMember(document, keyLine, key, want) + } + + items := document.sequenceItems(keyLine, childIndent) + if len(items) == 0 && document.lastContent(keyLine+1, document.blockEnd(childIndent, keyLine+1)) > keyLine { + return nil, false, fmt.Errorf("plugins.%s must be a list", key) + } + existing := -1 + for _, item := range items { + if value, _, _ := yamlSequenceItem(document.lines[item]); value == hermesPluginName { + existing = item + break + } + } + + if !want { + if existing < 0 { + return data, false, nil + } + document.remove(existing) + return document.bytes(), true, nil + } + if existing >= 0 { + return data, false, nil + } + itemIndent, insertAt := childIndent+2, keyLine+1 + if len(items) > 0 { + last := items[len(items)-1] + _, itemIndent, _ = yamlSequenceItem(document.lines[last]) + // An item may span more than its first line, so append after the last + // line that still belongs to it rather than after its opening dash. + insertAt = document.lastContent(last, document.blockEnd(itemIndent, last+1)) + 1 + } + document.insert(insertAt, strings.Repeat(" ", itemIndent)+"- "+hermesPluginName) + return document.bytes(), true, nil +} + +// setHermesPluginFlowMember edits an inline "key: [a, b]" list without parsing +// or rewriting its existing values. Patch appends its own scalar at the end; +// Unpatch removes only that exact trailing scalar. If an operator moves or +// rewrites the entry, Unpatch reports the conflict instead of risking the rest +// of the flow sequence. +func setHermesPluginFlowMember(document *yamlLines, keyLine int, key string, want bool) ([]byte, bool, error) { + line := document.lines[keyLine] + open := strings.IndexByte(line, '[') + end := strings.LastIndexByte(line, ']') + if open < 0 || end < open { + return nil, false, fmt.Errorf("plugins.%s must be a list", key) + } + + enabled, disabled, err := hermesPluginMembershipOf(document.bytes()) + if err != nil { + return nil, false, err + } + found := enabled + if key == "disabled" { + found = disabled + } + if found == want { + return document.bytes(), false, nil + } + + content := line[open+1 : end] + core := strings.TrimRight(content, " \t") + trailing := content[len(core):] + if want { + separator := "" + if strings.TrimSpace(core) != "" { + separator = ", " + } + document.lines[keyLine] = line[:open+1] + core + separator + hermesPluginName + trailing + line[end:] + return document.bytes(), true, nil + } + + if strings.TrimSpace(core) == hermesPluginName { + document.lines[keyLine] = line[:open+1] + trailing + line[end:] + return document.bytes(), true, nil + } + separator := strings.LastIndexByte(core, ',') + if separator < 0 || strings.TrimSpace(core[separator+1:]) != hermesPluginName { + return nil, false, fmt.Errorf("plugins.%s Gateway entry is not the final flow-list item", key) + } + document.lines[keyLine] = line[:open+1] + core[:separator] + trailing + line[end:] + return document.bytes(), true, nil +} + +// yamlLines is a line-oriented view of a YAML document. It understands just +// enough of the syntax to find a mapping key and the block sequence under it; +// every shape beyond that it declines to touch, leaving the caller to report a +// config it should not be editing blind. +type yamlLines struct { + lines []string + newline string +} + +func newYAMLLines(data []byte) *yamlLines { + newline := "\n" + if bytes.Contains(data, []byte("\r\n")) { + newline = "\r\n" + } + text := strings.ReplaceAll(string(data), "\r\n", "\n") + return &yamlLines{lines: strings.Split(text, "\n"), newline: newline} +} + +func (y *yamlLines) bytes() []byte { + return []byte(strings.Join(y.lines, y.newline)) +} + +func (y *yamlLines) insert(at int, lines ...string) { + y.lines = append(y.lines[:at:at], append(lines, y.lines[at:]...)...) +} + +func (y *yamlLines) remove(at int) { + y.lines = append(y.lines[:at:at], y.lines[at+1:]...) +} + +// appendBlock adds lines at the end of the document, keeping the file's final +// newline and separating the new block from whatever content came before it. +func (y *yamlLines) appendBlock(lines ...string) { + at := len(y.lines) + if at > 0 && y.lines[at-1] == "" { + // The empty last element stands for the file's final newline. + at-- + } else { + y.lines = append(y.lines, "") + } + if at > 0 && strings.TrimSpace(y.lines[at-1]) != "" { + lines = append([]string{""}, lines...) + } + y.insert(at, lines...) +} + +// findMappingKey returns the line index of key at indent inside [start, end), +// or -1. The scan stops at the first content line shallower than indent, which +// is where the surrounding block ends. +func (y *yamlLines) findMappingKey(key string, indent, start, end int) int { + for index := start; index < end && index < len(y.lines); index++ { + current, ok := yamlContentIndent(y.lines[index]) + if !ok || current > indent { + continue + } + if current < indent { + return -1 + } + if name, _, ok := yamlMappingKey(y.lines[index], indent); ok && name == key { + return index + } + } + return -1 +} + +// blockEnd returns the first line index at or after start that leaves the block +// owned by a key at indent. +func (y *yamlLines) blockEnd(indent, start int) int { + for index := start; index < len(y.lines); index++ { + if current, ok := yamlContentIndent(y.lines[index]); ok && current <= indent { + return index + } + } + return len(y.lines) +} + +// childIndent is the indentation of the first content line inside the block +// under keyLine, or two past the key's own indent when the block is empty. +func (y *yamlLines) childIndent(keyLine, indent, end int) int { + for index := keyLine + 1; index < end && index < len(y.lines); index++ { + if current, ok := yamlContentIndent(y.lines[index]); ok { + return current + } + } + return indent + 2 +} + +// lastContent is the index of the last content line in [start, end), or +// start-1 when the range holds none, so callers can insert right after it. +func (y *yamlLines) lastContent(start, end int) int { + last := start - 1 + for index := start; index < end && index < len(y.lines); index++ { + if _, ok := yamlContentIndent(y.lines[index]); ok { + last = index + } + } + return last +} + +// sequenceItems returns the line index of every item of the block sequence +// under the key at keyLine. Items may be indented deeper than their key or, as +// YAML also allows, sit at the key's own indentation. +func (y *yamlLines) sequenceItems(keyLine, keyIndent int) []int { + var items []int + for index := keyLine + 1; index < len(y.lines); index++ { + current, ok := yamlContentIndent(y.lines[index]) + if !ok { + continue + } + if _, itemIndent, isItem := yamlSequenceItem(y.lines[index]); isItem && itemIndent >= keyIndent { + items = append(items, index) + continue + } + // Anything deeper is a continuation of the item above it; anything at + // or above the key's own indent ends the sequence. + if current <= keyIndent { + break + } + } + return items +} + +// yamlContentIndent returns the indentation of a line that carries content. +// Blank lines and whole-line comments belong to whichever block surrounds them, +// so they report no content and every scan above skips over them. +func yamlContentIndent(line string) (int, bool) { + trimmed := strings.TrimLeft(line, " ") + if trimmed == "" || strings.HasPrefix(trimmed, "#") { + return 0, false + } + return len(line) - len(trimmed), true +} + +// yamlMappingKey reports the key and the inline value of a "<key>: <value>" +// line sitting at exactly indent. Only plain unquoted keys are recognized, +// which covers every key Gateway looks for. +func yamlMappingKey(line string, indent int) (string, string, bool) { + current, ok := yamlContentIndent(line) + if !ok || current != indent { + return "", "", false + } + key, value, ok := strings.Cut(line[indent:], ":") + if !ok || !isPlainYAMLKey(key) { + return "", "", false + } + return key, yamlInlineValue(value), true +} + +func isPlainYAMLKey(key string) bool { + if key == "" { + return false + } + for _, char := range key { + switch { + case char >= 'a' && char <= 'z', char >= 'A' && char <= 'Z', + char >= '0' && char <= '9', char == '_', char == '-', char == '.': + default: + return false + } + } + return true +} + +// yamlInlineValue strips the trailing comment from the value half of a mapping +// line. A flow sequence ends at its closing bracket; anything else ends where a +// comment starts. +func yamlInlineValue(value string) string { + value = strings.TrimSpace(value) + if strings.HasPrefix(value, "#") { + return "" + } + if strings.HasPrefix(value, "[") { + if end := strings.LastIndex(value, "]"); end >= 0 { + return value[:end+1] + } + return value + } + if comment := strings.Index(value, " #"); comment >= 0 { + return strings.TrimSpace(value[:comment]) + } + return value +} + +// yamlSequenceItem reports the scalar of a "- value" line and its indent. +// Quotes are stripped so a quoted plugin name matches the plain one Gateway +// writes. +func yamlSequenceItem(line string) (string, int, bool) { + indent, ok := yamlContentIndent(line) + if !ok { + return "", 0, false + } + rest := line[indent:] + if rest != "-" && !strings.HasPrefix(rest, "- ") { + return "", 0, false + } + return strings.Trim(yamlInlineValue(rest[1:]), `"'`), indent, true +} + +// hermesPluginMembership reports whether config.yaml lists the observer in +// plugins.enabled and in plugins.disabled. A missing config lists neither. +func hermesPluginMembership(configPath string) (bool, bool, error) { + data, err := os.ReadFile(configPath) + if os.IsNotExist(err) { + return false, false, nil + } + if err != nil { + return false, false, err + } + return hermesPluginMembershipOf(data) +} + +func hermesPluginMembershipOf(data []byte) (bool, bool, error) { + root, err := parseYAMLMapping(data) + if err != nil { + return false, false, err + } + plugins, ok := mappingValue(root, "plugins") + if !ok { + return false, false, nil + } + if plugins.Kind != yaml.MappingNode { + return false, false, errors.New("plugins must be a mapping") + } + enabled, err := pluginSequence(plugins, "enabled") + if err != nil { + return false, false, err + } + disabled, err := pluginSequence(plugins, "disabled") + if err != nil { + return false, false, err + } + return sequenceContains(enabled, hermesPluginName), + sequenceContains(disabled, hermesPluginName), nil +} + +// pluginSequence returns the plugins.<key> list, or nil when the key is absent +// or empty. Removing the last entry leaves the key behind with nothing under +// it, which Hermes reads the same way it reads a missing key, so an empty list +// written that way is not an error here either. +func pluginSequence(plugins *yaml.Node, key string) (*yaml.Node, error) { + value, ok := mappingValue(plugins, key) + if !ok || value.Tag == "!!null" { + return nil, nil + } + if value.Kind != yaml.SequenceNode { + return nil, fmt.Errorf("plugins.%s must be a sequence", key) + } + return value, nil +} + +func parseYAMLMapping(data []byte) (*yaml.Node, error) { + if len(bytes.TrimSpace(data)) == 0 { + return &yaml.Node{Kind: yaml.MappingNode, Tag: "!!map"}, nil + } + var document yaml.Node + if err := yaml.Unmarshal(data, &document); err != nil { + return nil, err + } + if len(document.Content) != 1 || document.Content[0].Kind != yaml.MappingNode { + return nil, errors.New("config root must be a mapping") + } + return document.Content[0], nil +} + +func mappingValue(mapping *yaml.Node, key string) (*yaml.Node, bool) { + if mapping == nil || mapping.Kind != yaml.MappingNode { + return nil, false + } + for index := 0; index+1 < len(mapping.Content); index += 2 { + if mapping.Content[index].Value == key { + return mapping.Content[index+1], true + } + } + return nil, false +} + +func sequenceContains(sequence *yaml.Node, value string) bool { + if sequence == nil || sequence.Kind != yaml.SequenceNode { + return false + } + for _, item := range sequence.Content { + if item.Kind == yaml.ScalarNode && item.Value == value { + return true + } + } + return false +} diff --git a/agentpatch/hermes_observer.py b/agentpatch/hermes_observer.py new file mode 100644 index 0000000..efe3bed --- /dev/null +++ b/agentpatch/hermes_observer.py @@ -0,0 +1,530 @@ +"""Casbin Gateway's behaviour observer for Hermes Agent. + +This module intentionally never serializes prompts, responses, reasoning, +tool results, or subagent text. Tool arguments are redacted and bounded before +hook callbacks enqueue records without waiting for the Gateway HTTP endpoint. +""" + +from __future__ import annotations + +import atexit +import json +import logging +import queue +import re +import threading +import urllib.request +from pathlib import Path +from typing import Any + + +_LOGGER = logging.getLogger(__name__) +_CONFIG_PATH = Path(__file__).with_name("gateway.json") +_QUEUE: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=1024) +_STOP = threading.Event() +_SENDER: threading.Thread | None = None +_REGISTER_LOCK = threading.Lock() +_MAX_OBJECT_BYTES = 64 * 1024 + +_CREDENTIAL_PATTERN = re.compile( + r"\b(?:sk-(?:ant-|proj-)?[a-z0-9_-]{12,}|gh[pousr]_[a-z0-9]{20,}|" + r"github_pat_[a-z0-9_]{20,}|AKIA[0-9A-Z]{16}|AIza[0-9a-z_-]{30,}|" + r"xox[baprs]-[0-9a-z-]{12,}|eyJ[a-z0-9_-]{10,}\.[a-z0-9_-]{10,}\." + r"[a-z0-9_-]{10,})\b", + re.IGNORECASE, +) +_BEARER_PATTERN = re.compile(r"(bearer\s+)[a-z0-9._~+/=-]{12,}", re.IGNORECASE) +_PRIVATE_KEY_PATTERN = re.compile( + r"-----BEGIN [^-\n]*PRIVATE KEY-----.*?-----END [^-\n]*PRIVATE KEY-----", + re.DOTALL, +) +_SENSITIVE_KEY_MARKERS = ( + "secret", + "token", + "password", + "passwd", + "credential", + "privatekey", + "apikey", + "accesskey", + "authorization", + "cookie", +) + +_TOKEN_COUNT_KEYS = { + "maxtokens", + "maxcompletiontokens", + "maxtokenstosample", + "maxoutputtokens", + "prompttokens", + "completiontokens", + "totaltokens", + "reasoningtokens", + "inputtokens", + "outputtokens", + "tokencount", + "numtokens", + "tokens", + "cachecreationinputtokens", + "cachereadinputtokens", + "cachedtokens", +} + +_USAGE_KEYS = { + "input_tokens", + "output_tokens", + "total_tokens", + "prompt_tokens", + "completion_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", + "reasoning_tokens", + "inputTokens", + "outputTokens", + "totalTokens", + "promptTokens", + "completionTokens", + "cacheReadInputTokens", + "cacheCreationInputTokens", + "reasoningTokens", +} + + +def _load_config() -> dict[str, Any]: + try: + value = json.loads(_CONFIG_PATH.read_text(encoding="utf-8")) + return value if isinstance(value, dict) else {} + except Exception: + return {} + + +_CONFIG = _load_config() +_RECORDS_URL = str(_CONFIG.get("recordsUrl") or "") +_AGENT_PATH = str(_CONFIG.get("agentPath") or "") +_USER = str(_CONFIG.get("user") or "") +_INGEST_TOKEN = str(_CONFIG.get("ingestToken") or "") +_INGEST_TOKEN_HEADER = str(_CONFIG.get("ingestTokenHeader") or "") + + +def _text(value: Any, limit: int = 512) -> str: + if value is None: + return "" + if isinstance(value, (str, int, float, bool)): + return str(value)[:limit] + return "" + + +def _integer(value: Any) -> int | None: + if isinstance(value, bool): + return None + if isinstance(value, (int, float)): + return int(value) + return None + + +def _duration_ms(value: Any, *, seconds: bool = False) -> int: + try: + number = float(value) + except (TypeError, ValueError): + return 0 + if seconds: + number *= 1000 + return max(0, int(number)) + + +def _length(value: Any) -> int: + if value is None: + return 0 + try: + return len(value) + except (TypeError, ValueError): + return len(str(value)) + + +def _count(value: Any) -> int: + if isinstance(value, (list, tuple, set, dict)): + return len(value) + return 0 + + +def _metadata(**values: Any) -> dict[str, Any]: + return {key: value for key, value in values.items() if value not in ("", None)} + + +def _sanitize_value(key: str, value: Any) -> Any: + normalized_key = key.lower().replace("_", "").replace("-", "").replace(".", "") + if normalized_key not in _TOKEN_COUNT_KEYS and any( + marker in normalized_key for marker in _SENSITIVE_KEY_MARKERS + ): + return "[REDACTED]" + if isinstance(value, dict): + return {child_key: _sanitize_value(child_key, child) for child_key, child in value.items()} + if isinstance(value, list): + return [_sanitize_value("", child) for child in value] + if isinstance(value, str): + value = _PRIVATE_KEY_PATTERN.sub("[REDACTED PRIVATE KEY]", value) + value = _BEARER_PATTERN.sub(r"\1[REDACTED]", value) + return _CREDENTIAL_PATTERN.sub("[REDACTED]", value) + return value + + +def _sensitive_path(file_path: str) -> bool: + normalized = file_path.lower().replace("\\", "/") + base = normalized.rsplit("/", 1)[-1] + if base.startswith(".env") and base not in {".env.example", ".env.sample", ".env.template"}: + return True + if normalized.startswith(".ssh/") or "/.ssh/" in normalized: + return True + if normalized == ".aws/credentials" or normalized.endswith("/.aws/credentials"): + return True + if base in {".npmrc", ".pypirc", "credentials", "id_rsa", "id_ed25519"}: + return True + return base.endswith((".pem", ".key")) + + +def _sanitize_tool_arguments(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]: + sanitized = _sanitize_value("", arguments) + normalized_tool = tool_name.lower() + if (normalized_tool == "patch" or normalized_tool.endswith("__patch")) and "patch" in sanitized: + sanitized["patch"] = "[OMITTED: patch content]" + sensitive_write = normalized_tool in {"write", "edit", "write_file", "edit_file", "patch"} or normalized_tool.endswith( + ("__write_file", "__edit_file", "__patch") + ) + file_path = arguments.get("path", "") or arguments.get("file_path", "") + if not sensitive_write or not _sensitive_path(file_path): + return sanitized + for key in ("content", "old_string", "new_string"): + if key in sanitized: + sanitized[key] = "[REDACTED: sensitive file content]" + return sanitized + + +def _usage(value: Any) -> dict[str, int]: + if not isinstance(value, dict): + return {} + result: dict[str, int] = {} + for key in _USAGE_KEYS: + number = _integer(value.get(key)) + if number is not None: + result[key] = number + return result + + +def _mcp_identity(tool_name: str) -> tuple[str, str]: + parts = tool_name.split("__", 2) + if len(parts) == 3 and parts[0].lower() == "mcp" and parts[1] and parts[2]: + return parts[1], parts[2] + return "", "" + + +def _base_record( + event_type: str, + action: str, + outcome: str, + values: dict[str, Any], + metadata: dict[str, Any] | None = None, +) -> dict[str, Any]: + record: dict[str, Any] = { + "agent": "hermes-agent", + "agentPath": _AGENT_PATH, + "user": _USER, + "eventType": event_type, + "action": action, + "sessionKey": _text( + values.get("session_id") + or values.get("parent_session_id") + or values.get("child_session_id") + ), + "model": _text(values.get("model")), + "promptId": _text( + values.get("api_request_id") + or values.get("turn_id") + or values.get("task_id") + ), + } + if outcome: + record["outcome"] = outcome + if metadata: + encoded = json.dumps( + metadata, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ) + encoded_bytes = encoded.encode("utf-8") + if len(encoded_bytes) > _MAX_OBJECT_BYTES: + encoded = json.dumps( + { + "truncated": True, + "originalBytes": len(encoded_bytes), + "preview": encoded_bytes[: _MAX_OBJECT_BYTES // 3].decode( + "utf-8", errors="ignore" + ), + }, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + record["object"] = encoded + return {key: value for key, value in record.items() if value not in ("", None)} + + +def _enqueue(record: dict[str, Any]) -> None: + if not _RECORDS_URL: + return + try: + _QUEUE.put_nowait(record) + except queue.Full: + _LOGGER.debug("Gateway observer queue is full; dropping one audit record") + + +def _send(record: dict[str, Any]) -> None: + body = json.dumps(record, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + headers = {"Content-Type": "application/json"} + if _INGEST_TOKEN_HEADER and _INGEST_TOKEN: + headers[_INGEST_TOKEN_HEADER] = _INGEST_TOKEN + request = urllib.request.Request( + _RECORDS_URL, + data=body, + headers=headers, + method="POST", + ) + with urllib.request.urlopen(request, timeout=2.0) as response: + response.read(1) + + +def _sender_loop() -> None: + while not _STOP.is_set() or not _QUEUE.empty(): + try: + record = _QUEUE.get(timeout=0.2) + except queue.Empty: + continue + try: + _send(record) + except Exception as exc: + _LOGGER.debug("Gateway record delivery failed: %s", type(exc).__name__) + finally: + _QUEUE.task_done() + + +def _start_sender() -> None: + global _SENDER + if not _RECORDS_URL: + return + with _REGISTER_LOCK: + if _SENDER is not None and _SENDER.is_alive(): + return + _SENDER = threading.Thread( + target=_sender_loop, + name="casbin-gateway-agent-monitor", + daemon=True, + ) + _SENDER.start() + + +def _shutdown() -> None: + _STOP.set() + sender = _SENDER + if sender is not None and sender.is_alive(): + sender.join(timeout=1.0) + + +atexit.register(_shutdown) + + +def _on_pre_api_request(**values: Any) -> None: + data = _metadata( + provider=_text(values.get("provider")), + apiMode=_text(values.get("api_mode")), + apiCallCount=_integer(values.get("api_call_count")), + retryCount=_integer(values.get("retry_count")), + messageCount=_integer(values.get("message_count")), + toolCount=_integer(values.get("tool_count")), + approxInputTokens=_integer(values.get("approx_input_tokens")), + requestCharCount=_integer(values.get("request_char_count")), + maxTokens=_integer(values.get("max_tokens")), + ) + _enqueue(_base_record("llm", "request", "attempted", values, data)) + + +def _on_post_api_request(**values: Any) -> None: + data = _metadata( + provider=_text(values.get("provider")), + apiMode=_text(values.get("api_mode")), + apiCallCount=_integer(values.get("api_call_count")), + retryCount=_integer(values.get("retry_count")), + finishReason=_text(values.get("finish_reason")), + responseModel=_text(values.get("response_model")), + messageCount=_integer(values.get("message_count")), + assistantContentChars=_integer(values.get("assistant_content_chars")), + assistantToolCallCount=_integer(values.get("assistant_tool_call_count")), + usage=_usage(values.get("usage")), + ) + record = _base_record("llm", "response", "success", values, data) + record["durationMs"] = _duration_ms(values.get("api_duration"), seconds=True) + _enqueue(record) + + +def _error_identity(values: dict[str, Any]) -> tuple[str, Any]: + """Split Hermes' error report into a type name and the message itself. + + ``api_request_error`` reports a structured ``error = {"type", "message"}``; + the tool hooks report flat ``error_type`` / ``error_message`` instead. Only + the type name and the message's length ever leave this module. + """ + error = values.get("error") + error_type = _text(values.get("error_type")) + message = values.get("error_message") + if isinstance(error, dict): + error_type = error_type or _text(error.get("type")) + if message is None: + message = error.get("message") + elif message is None and error is not None: + message = error + if not error_type and message is not None: + error_type = type(message).__name__ + return error_type, message + + +def _on_api_request_error(**values: Any) -> None: + error_type, error_message = _error_identity(values) + data = _metadata( + provider=_text(values.get("provider")), + apiMode=_text(values.get("api_mode")), + apiCallCount=_integer(values.get("api_call_count")), + retryCount=_integer(values.get("retry_count")), + maxRetries=_integer(values.get("max_retries")), + retryable=values.get("retryable") if isinstance(values.get("retryable"), bool) else None, + statusCode=_integer(values.get("status_code")), + reason=_text(values.get("reason")), + errorType=error_type, + errorMessageLength=_length(error_message) if error_message is not None else None, + ) + record = _base_record("llm", "error", "failure", values, data) + if values.get("api_duration") is not None: + record["durationMs"] = _duration_ms(values.get("api_duration"), seconds=True) + _enqueue(record) + + +def _tool_record(action: str, outcome: str, values: dict[str, Any]) -> dict[str, Any]: + tool_name = _text(values.get("tool_name")) + server, tool = _mcp_identity(tool_name) + event_type = "mcp" if server else "tool" + error_type, error_message = _error_identity(values) + data = _metadata( + taskId=_text(values.get("task_id")), + turnId=_text(values.get("turn_id")), + errorType=error_type, + errorMessageLength=_length(error_message) if error_message is not None else None, + middlewareStageCount=_count(values.get("middleware_trace")), + ) + if action == "call": + data["arguments"] = _sanitize_tool_arguments(tool_name, values["args"]) + record = _base_record(event_type, action, outcome, values, data) + record["toolName"] = tool_name + record["toolUseId"] = _text(values.get("tool_call_id")) + if server: + record["mcpServer"] = server + record["mcpTool"] = tool + duration = _duration_ms(values.get("duration_ms")) + if duration: + record["durationMs"] = duration + return {key: value for key, value in record.items() if value not in ("", None)} + + +def _on_pre_tool_call(**values: Any) -> None: + _enqueue(_tool_record("call", "attempted", values)) + + +def _on_post_tool_call(**values: Any) -> None: + status = _text(values.get("status")).lower() + outcome = {"ok": "success", "blocked": "denied", "error": "failure"}.get( + status, "failure" + ) + _enqueue(_tool_record("result", outcome, values)) + + +def _on_session_start(**values: Any) -> None: + data = _metadata(taskId=_text(values.get("task_id"))) + _enqueue(_base_record("session", "start", "", values, data)) + + +def _on_session_reset(**values: Any) -> None: + normalized = dict(values) + normalized["session_id"] = values.get("new_session_id") or values.get("session_id") + data = _metadata( + oldSessionId=_text(values.get("old_session_id")), + reason=_text(values.get("reason")), + ) + _enqueue(_base_record("session", "reset", "success", normalized, data)) + + +def _on_session_end(**values: Any) -> None: + failed = bool(values.get("failed")) + interrupted = bool(values.get("interrupted")) + outcome = "failure" if failed or interrupted else "success" + data = _metadata( + taskId=_text(values.get("task_id")), + turnId=_text(values.get("turn_id")), + completed=bool(values.get("completed")), + failed=failed, + interrupted=interrupted, + exitReason=_text(values.get("turn_exit_reason")), + ) + _enqueue(_base_record("session", "end", outcome, values, data)) + + +def _on_subagent_start(**values: Any) -> None: + normalized = dict(values) + normalized["session_id"] = values.get("parent_session_id") or values.get("session_id") + normalized["turn_id"] = values.get("parent_turn_id") or values.get("turn_id") + data = _metadata( + subagentId=_text(values.get("child_subagent_id") or values.get("subagent_id")), + childSessionId=_text(values.get("child_session_id")), + childRole=_text(values.get("child_role")), + goalLength=_length(values.get("child_goal")), + ) + _enqueue(_base_record("subagent", "start", "attempted", normalized, data)) + + +def _on_subagent_stop(**values: Any) -> None: + normalized = dict(values) + normalized["session_id"] = values.get("parent_session_id") or values.get("session_id") + normalized["turn_id"] = values.get("parent_turn_id") or values.get("turn_id") + status = _text(values.get("child_status") or values.get("status")).lower() + outcome = "success" if status in {"ok", "success", "completed"} else "failure" + data = _metadata( + subagentId=_text(values.get("child_subagent_id") or values.get("subagent_id")), + childSessionId=_text(values.get("child_session_id")), + childRole=_text(values.get("child_role")), + status=status, + summaryLength=_length(values.get("child_summary")), + toolCallCount=_count(values.get("tool_call_history")), + ) + # "stop", not "end": the rest of Gateway's record vocabulary already spells + # the end of a subagent that way (see agenthook's SubagentStop). + record = _base_record("subagent", "stop", outcome, normalized, data) + duration = values.get("duration_ms") + if duration is None and values.get("duration") is not None: + record["durationMs"] = _duration_ms(values.get("duration"), seconds=True) + elif duration is not None: + record["durationMs"] = _duration_ms(duration) + _enqueue(record) + + +_HOOKS = { + "pre_api_request": _on_pre_api_request, + "post_api_request": _on_post_api_request, + "api_request_error": _on_api_request_error, + "pre_tool_call": _on_pre_tool_call, + "post_tool_call": _on_post_tool_call, + "on_session_start": _on_session_start, + "on_session_end": _on_session_end, + "on_session_reset": _on_session_reset, + "subagent_start": _on_subagent_start, + "subagent_stop": _on_subagent_stop, +} + + +def register(ctx: Any) -> None: + """Register observational hooks. Return values are deliberately ignored.""" + _start_sender() + for name, callback in _HOOKS.items(): + ctx.register_hook(name, callback) diff --git a/agentpatch/hermes_plugin.yaml b/agentpatch/hermes_plugin.yaml new file mode 100644 index 0000000..83951cc --- /dev/null +++ b/agentpatch/hermes_plugin.yaml @@ -0,0 +1,18 @@ +name: casbin-gateway-agent-monitor +version: "1.1.0" +description: "Behaviour audit with redacted tool inputs for Casbin Gateway." +author: Casbin +kind: standalone +# "provides_hooks" is the key Hermes' manifest parser reads; the callbacks +# themselves are registered from register() in __init__.py. +provides_hooks: + - pre_api_request + - post_api_request + - api_request_error + - pre_tool_call + - post_tool_call + - on_session_start + - on_session_end + - on_session_reset + - subagent_start + - subagent_stop diff --git a/go.mod b/go.mod index 0bc0062..b27972b 100644 --- a/go.mod +++ b/go.mod @@ -27,6 +27,7 @@ require ( golang.org/x/sys v0.17.0 golang.org/x/text v0.14.0 golang.org/x/time v0.3.0 + gopkg.in/yaml.v3 v3.0.1 modernc.org/sqlite v1.18.1 ) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
