cli: expand 'show nacm [user NAME | group NAME]'

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>

src/bin/show/__init__.py

@@ -11,21 +11,26 @@ import argparse
 RAW_OUTPUT = False
 
 
-def get_json(xpath: str, datastore: str = "operational") -> dict:
+def get_json(xpath: str, datastore: str = "operational", quiet: bool = False) -> dict:
     if not isinstance(xpath, str) or not xpath.startswith("/"):
-        print("Invalid XPATH. It must be a valid string starting with '/'.")
+        if not quiet:
+            print("Invalid XPATH. It must be a valid string starting with '/'.")
         return {}
 
     try:
         result = subprocess.run(["copy", datastore, "-x", shlex.quote(xpath)],
                                 capture_output=True, text=True, check=True)
+        if not result.stdout.strip():
+            return {}
         json_data = json.loads(result.stdout)
         return json_data
     except subprocess.CalledProcessError as e:
-        print(f"Error running copy: {e}")
+        if not quiet:
+            print(f"Error running copy: {e}")
         return {}
     except json.JSONDecodeError as e:
-        print(f"Error parsing JSON output: {e}")
+        if not quiet:
+            print(f"Error parsing JSON output: {e}")
         return {}
 
 
@@ -606,13 +611,21 @@ def system(args: List[str]) -> None:
 
 
 def nacm(args: List[str]) -> None:
-    data = get_json("/ietf-netconf-acm:nacm", "running")
+    """Handle show nacm [subcommand] [name]
+
+    Subcommands:
+        (none)      - Overview with users, groups, and permissions matrix
+        group NAME  - Detailed view of a specific group's rules
+        user NAME   - Detailed view of a specific user's permissions
+    """
+    data = get_json("/ietf-netconf-acm:nacm", "running", quiet=True)
     if not data:
-        print("No NACM data retrieved.")
+        user = os.environ.get('USER', 'unknown')
+        print(f'Error: Access denied because "{user}" NACM authorization failed.')
         return
 
     # Fetch user data from operational (includes shell and authorized-key from yanger)
-    user_oper = get_json("/ietf-system:system/authentication")
+    user_oper = get_json("/ietf-system:system/authentication", quiet=True)
 
     if user_oper:
         oper_users = user_oper.get("ietf-system:system", {}).get("authentication", {}).get("user", [])
@@ -621,7 +634,28 @@ def nacm(args: List[str]) -> None:
     if RAW_OUTPUT:
         print(json.dumps(data, indent=2))
         return
-    cli_pretty(data, "show-nacm")
+
+    # Parse arguments: subcommand and optional name
+    subcommand = args[0] if len(args) > 0 and args[0] else ""
+    name = args[1] if len(args) > 1 else None
+
+    # Route to appropriate formatter
+    if subcommand == "":
+        cli_pretty(data, "show-nacm")
+    elif subcommand == "group":
+        if not name:
+            print("Usage: show nacm group <name>")
+            return
+        data['_group_name'] = name
+        cli_pretty(data, "show-nacm-group")
+    elif subcommand == "user":
+        if not name:
+            print("Usage: show nacm user <name>")
+            return
+        data['_user_name'] = name
+        cli_pretty(data, "show-nacm-user")
+    else:
+        print(f"Unknown NACM subcommand: {subcommand}")
 
 
 def execute_command(command: str, args: List[str]):

src/klish-plugin-infix/src/infix.c

@@ -360,7 +360,7 @@ static const char *valid_boot_target(const kparg_t *parg)
 	return NULL;
 }
 
-int infix_set_boot_order(kcontext_t *ctx)
+static int infix_set_boot_order(kcontext_t *ctx)
 {
 	kpargv_t *pargv = kcontext_pargv(ctx);
 	const char *targets[3];
@@ -399,6 +399,20 @@ int infix_set_boot_order(kcontext_t *ctx)
 	return run(argv);
 }
 
+int infix_users(kcontext_t *ctx)
+{
+	(void)ctx;
+	return shellf("copy oper -x /system/authentication/user/name "
+		      "| jq -r '.\"ietf-system:system\".authentication.user[].name'");
+}
+
+int infix_groups(kcontext_t *ctx)
+{
+	(void)ctx;
+	return shellf("copy oper -x /nacm/groups "
+		      "| jq -r '.\"ietf-netconf-acm:nacm\".groups.group[].name'");
+}
+
 int kplugin_infix_fini(kcontext_t *ctx)
 {
 	(void)ctx;
@@ -416,6 +430,8 @@ int kplugin_infix_init(kcontext_t *ctx)
 	kplugin_add_syms(plugin, ksym_new("erase", infix_erase));
 	kplugin_add_syms(plugin, ksym_new("files", infix_files));
 	kplugin_add_syms(plugin, ksym_new("ifaces", infix_ifaces));
+	kplugin_add_syms(plugin, ksym_new("users", infix_users));
+	kplugin_add_syms(plugin, ksym_new("groups", infix_groups));
 	kplugin_add_syms(plugin, ksym_new("firewall_zones", infix_firewall_zones));
 	kplugin_add_syms(plugin, ksym_new("firewall_policies", infix_firewall_policies));
 	kplugin_add_syms(plugin, ksym_new("firewall_services", infix_firewall_services));

src/klish-plugin-infix/xml/infix.xml

@@ -141,6 +141,20 @@
   <ACTION sym="STRING"/>
 </PTYPE>
 
+<PTYPE name="USERS">
+  <COMPL>
+    <ACTION sym="users@infix"/>
+  </COMPL>
+  <ACTION sym="STRING"/>
+</PTYPE>
+
+<PTYPE name="GROUPS">
+  <COMPL>
+    <ACTION sym="groups@infix"/>
+  </COMPL>
+  <ACTION sym="STRING"/>
+</PTYPE>
+
 <VIEW name="main">
   <HOTKEY key="^D" cmd="exit"/>
 
@@ -325,6 +339,16 @@
   </COMMAND>
 
   <COMMAND name="nacm" help="Show users and NACM status and groups">
+    <SWITCH name="subcommands" min="0">
+      <COMMAND name="group" help="Show details for a specific NACM group">
+	<PARAM name="name" ptype="/GROUPS" help="Group name"/>
+        <ACTION sym="script" in="tty" out="tty" interrupt="true">show nacm group $KLISH_PARAM_name</ACTION>
+      </COMMAND>
+      <COMMAND name="user" help="Show NACM details for a specific user">
+	<PARAM name="name" ptype="/USERS" help="User name"/>
+        <ACTION sym="script" in="tty" out="tty" interrupt="true">show nacm user $KLISH_PARAM_name</ACTION>
+      </COMMAND>
+    </SWITCH>
     <ACTION sym="script" in="tty" out="tty" interrupt="true">show nacm</ACTION>
   </COMMAND>
 

src/statd/python/cli_pretty/cli_pretty.py

@@ -2906,6 +2906,198 @@ def show_ntp_source(json, address=None):
         print(row)
 
 
+def _analyze_group_permissions(nacm):
+    """Analyze NACM rules to determine effective permissions per group.
+
+    NACM rule evaluation order:
+    1. Rule-lists are processed sequentially in configuration order
+    2. Within each rule-list, rules are evaluated sequentially
+    3. First matching rule wins - no further rules are evaluated
+    4. If no rule matches, global defaults apply
+    5. permit-all (module-name=*, access-operations=*) bypasses everything
+    """
+    read_default = nacm.get('read-default', 'permit') == 'permit'
+    write_default = nacm.get('write-default', 'permit') == 'permit'
+    exec_default = nacm.get('exec-default', 'permit') == 'permit'
+
+    rule_lists = nacm.get('rule-list', [])
+    groups_config = nacm.get('groups', {}).get('group', [])
+
+    # Collect all deny rules that apply to "*" (all groups)
+    # These create restrictions for groups without permit-all
+    global_denials = []
+    for rule_list in rule_lists:
+        if '*' in rule_list.get('group', []):
+            for rule in rule_list.get('rule', []):
+                if rule.get('action') == 'deny':
+                    global_denials.append(rule)
+
+    results = []
+
+    for group in groups_config:
+        group_name = group.get('name', '')
+
+        # Defaults
+        can_read = read_default
+        can_write = write_default
+        can_exec = exec_default
+        restrictions = []
+        has_permit_all = False
+
+        # Process rule-lists in order (first match wins)
+        for rule_list in rule_lists:
+            list_groups = rule_list.get('group', [])
+
+            # Check if this rule-list applies to this group specifically
+            if group_name not in list_groups:
+                continue
+
+            for rule in rule_list.get('rule', []):
+                action = rule.get('action')
+                access_ops = rule.get('access-operations', '')
+                module_name = rule.get('module-name', '')
+
+                # permit-all pattern: bypasses ALL other rules including global denials
+                if action == 'permit' and module_name == '*' and access_ops == '*':
+                    has_permit_all = True
+                    break
+
+                # deny pattern for write+exec (like guest)
+                if action == 'deny' and module_name == '*':
+                    ops = access_ops.lower()
+                    if all(op in ops for op in ['create', 'update', 'delete']):
+                        can_write = False
+                    if 'exec' in ops:
+                        can_exec = False
+
+            if has_permit_all:
+                break
+
+        # If not permit-all, global denials create restrictions
+        # These affect READ too (access-operations: "*" includes read)
+        if not has_permit_all:
+            for rule in global_denials:
+                path = rule.get('path', '')
+                module = rule.get('module-name', '')
+
+                if path:
+                    # Extract meaningful part: /ietf-system:system/.../password -> password
+                    restriction = path.rstrip('/').split('/')[-1]
+                elif module:
+                    # Remove ietf- prefix for brevity: ietf-keystore -> keystore
+                    restriction = module.replace('ietf-', '')
+                else:
+                    continue
+
+                if restriction and restriction not in restrictions:
+                    restrictions.append(restriction)
+
+        results.append({
+            'group': group_name,
+            'read': can_read,
+            'write': can_write,
+            'exec': can_exec,
+            'restrictions': restrictions,
+            'has_permit_all': has_permit_all
+        })
+
+    return results
+
+
+def _nacm_permissions_matrix(nacm, width=None):
+    """Render NACM group permissions as a colored matrix.
+
+    Symbols: ✓=full access, ⚠=restricted, ✗=denied
+
+    Returns:
+        Multi-line string containing the rendered matrix, or None if no groups
+    """
+    permissions = _analyze_group_permissions(nacm)
+    if not permissions:
+        return None
+
+    # Column widths
+    group_col_width = max(len(p['group']) for p in permissions)
+    group_col_width = max(group_col_width, 5)  # Minimum for "GROUP"
+    cell_width = 7  # Width for READ/WRITE/EXEC columns
+
+    # Box drawing
+    top_border = "┌" + "─" * (group_col_width + 2) + "┬"
+    top_border += "─" * (cell_width + 2) + "┬"
+    top_border += "─" * (cell_width + 2) + "┬"
+    top_border += "─" * (cell_width + 2) + "┐"
+
+    header_border = "├" + "─" * (group_col_width + 2) + "┼"
+    header_border += "─" * (cell_width + 2) + "┼"
+    header_border += "─" * (cell_width + 2) + "┼"
+    header_border += "─" * (cell_width + 2) + "┤"
+
+    bottom_border = "└" + "─" * (group_col_width + 2) + "┴"
+    bottom_border += "─" * (cell_width + 2) + "┴"
+    bottom_border += "─" * (cell_width + 2) + "┴"
+    bottom_border += "─" * (cell_width + 2) + "┘"
+
+    # Header row
+    header = f"│ {'GROUP':<{group_col_width}} │"
+    header += f" {'READ':^{cell_width}} │"
+    header += f" {'WRITE':^{cell_width}} │"
+    header += f" {'EXEC':^{cell_width}} │"
+
+    # Calculate centering
+    matrix_width = len(top_border)
+    if width and width > matrix_width:
+        padding = (width - matrix_width) // 2
+        indent = " " * padding
+    else:
+        indent = ""
+
+    lines = []
+    lines.append(indent + top_border)
+    lines.append(indent + header)
+    lines.append(indent + header_border)
+
+    # Data rows
+    for perm in permissions:
+        group_name = perm['group']
+        has_restrictions = bool(perm['restrictions'])
+
+        # Determine symbols and colors for each cell
+        def get_cell(has_access, has_restrictions):
+            if not has_access:
+                return Decore.red_bg(f" {'✗':^{cell_width}} ")
+            elif has_restrictions:
+                return Decore.yellow_bg(f" {'⚠':^{cell_width}} ")
+            else:
+                return Decore.green_bg(f" {'✓':^{cell_width}} ")
+
+        read_cell = get_cell(perm['read'], has_restrictions)
+        write_cell = get_cell(perm['write'], has_restrictions)
+        exec_cell = get_cell(perm['exec'], has_restrictions)
+
+        row = f"│ {group_name:<{group_col_width}} │{read_cell}│{write_cell}│{exec_cell}│"
+        lines.append(indent + row)
+
+    lines.append(indent + bottom_border)
+
+    # Legend
+    legend_data = [
+        ("✓ Full", Decore.green_bg),
+        ("⚠ Restricted", Decore.yellow_bg),
+        ("✗ Denied", Decore.red_bg)
+    ]
+    colorized_parts = [bg_func(f" {text} ") for text, bg_func in legend_data]
+    legend = "  ".join(colorized_parts)
+
+    if width:
+        visible_legend = "  ".join([f" {text} " for text, _ in legend_data])
+        legend_padding = max(0, (width - len(visible_legend)) // 2)
+        lines.append(" " * legend_padding + legend)
+    else:
+        lines.append(indent + legend)
+
+    return "\n".join(lines)
+
+
 def show_nacm(json):
     """Display users and NACM (Network Configuration Access Control) groups"""
     min_width = 62
@@ -2917,15 +3109,23 @@ def show_nacm(json):
 
     if nacm:
         enabled = "yes" if nacm.get("enable-nacm", True) else "no"
-        print(f"{'enabled':<25}: {enabled}")
-        print(f"{'default read access':<25}: {nacm.get('read-default', 'permit')}")
-        print(f"{'default write access':<25}: {nacm.get('write-default', 'deny')}")
-        print(f"{'default exec access':<25}: {nacm.get('exec-default', 'permit')}")
-        print(f"{'denied operations':<25}: {nacm.get('denied-operations', 0)}")
-        print(f"{'denied data writes':<25}: {nacm.get('denied-data-writes', 0)}")
-        print(f"{'denied notifications':<25}: {nacm.get('denied-notifications', 0)}")
+        print(f"{'enabled':<21}: {enabled}")
+        print(f"{'default read access':<21}: {nacm.get('read-default', 'permit')}")
+        print(f"{'default write access':<21}: {nacm.get('write-default', 'deny')}")
+        print(f"{'default exec access':<21}: {nacm.get('exec-default', 'permit')}")
+        print(f"{'denied operations':<21}: {nacm.get('denied-operations', 0)}")
+        print(f"{'denied data writes':<21}: {nacm.get('denied-data-writes', 0)}")
+        print(f"{'denied notifications':<21}: {nacm.get('denied-notifications', 0)}")
         print()
 
+
+    # Group permissions matrix
+    if nacm:
+        matrix = _nacm_permissions_matrix(nacm, min_width)
+        if matrix:
+            print(matrix)
+            print()
+
     # Users table
     system = json.get("ietf-system:system", {})
     users = system.get("authentication", {}).get("user", [])
@@ -2971,6 +3171,214 @@ def show_nacm(json):
             group_table.row(name, members)
 
         group_table.print()
+        print()
+
+
+def show_nacm_group(json):
+    """Display detailed information about a specific NACM group."""
+    group_name = json.get('_group_name')
+    nacm = json.get("ietf-netconf-acm:nacm", {})
+
+    if not nacm:
+        print("NACM not configured.")
+        return
+
+    # Find the group
+    groups = nacm.get("groups", {}).get("group", [])
+    group = None
+    for g in groups:
+        if g.get("name") == group_name:
+            group = g
+            break
+
+    if not group:
+        print(f"Group '{group_name}' not found.")
+        return
+
+    # Group info
+    members = group.get("user-name", [])
+    members_str = ", ".join(members) if members else "(none)"
+    print(f"{'members':<17}: {members_str}")
+
+    # Analyze permissions
+    permissions = _analyze_group_permissions(nacm)
+    group_perm = None
+    for p in permissions:
+        if p['group'] == group_name:
+            group_perm = p
+            break
+
+    if group_perm:
+        def perm_str(has_access, has_restrictions):
+            if not has_access:
+                return Decore.red("no")
+            elif has_restrictions:
+                return Decore.yellow("restricted")
+            else:
+                return Decore.green("yes")
+
+        has_restrictions = bool(group_perm['restrictions'])
+        print(f"{'read permission':<17}: {perm_str(group_perm['read'], has_restrictions)}")
+        print(f"{'write permission':<17}: {perm_str(group_perm['write'], has_restrictions)}")
+        print(f"{'exec permission':<17}: {perm_str(group_perm['exec'], has_restrictions)}")
+
+    # Find applicable rules, respecting NACM evaluation order
+    # Rule-lists are evaluated in order; first matching rule wins
+    # If a permit-all rule is found, no further rules would be evaluated
+    rule_lists = nacm.get("rule-list", [])
+    applicable_rules = []
+    has_permit_all = group_perm.get('has_permit_all', False) if group_perm else False
+
+    for rule_list in rule_lists:
+        list_groups = rule_list.get("group", [])
+
+        # Check if this rule-list applies to this group specifically (not via "*")
+        if group_name in list_groups:
+            list_name = rule_list.get("name", "")
+            for rule in rule_list.get("rule", []):
+                applicable_rules.append({
+                    'list': list_name,
+                    'rule': rule,
+                    'applies_via': group_name
+                })
+
+                # Check if this is a permit-all rule
+                action = rule.get('action')
+                module_name = rule.get('module-name', '')
+                access_ops = rule.get('access-operations', '')
+                if action == 'permit' and module_name == '*' and access_ops == '*':
+                    # permit-all matches everything, stop here
+                    break
+
+            if has_permit_all:
+                # Don't process any more rule-lists
+                break
+
+    # Only add rules from "*" group if we don't have permit-all
+    if not has_permit_all:
+        for rule_list in rule_lists:
+            list_groups = rule_list.get("group", [])
+            if "*" in list_groups and group_name not in list_groups:
+                list_name = rule_list.get("name", "")
+                for rule in rule_list.get("rule", []):
+                    applicable_rules.append({
+                        'list': list_name,
+                        'rule': rule,
+                        'applies_via': "*"
+                    })
+
+    if applicable_rules:
+        print(f"{'applicable rules':<17}: {len(applicable_rules)}")
+
+        for item in applicable_rules:
+            rule = item['rule']
+            rule_name = rule.get('name', '(unnamed)')
+            action = rule.get('action', 'permit')
+            access_ops = rule.get('access-operations', '*')
+            module_name = rule.get('module-name', '')
+            path = rule.get('path', '')
+            rpc_name = rule.get('rpc-name', '')
+            comment = rule.get('comment', '')
+
+            # Format action with color
+            if action == 'permit':
+                action_str = Decore.green("permit")
+            else:
+                action_str = Decore.red("deny")
+
+            # Build target description
+            target = ""
+            if path:
+                target = path
+            elif module_name:
+                target = module_name
+                if rpc_name:
+                    target += f" (rpc: {rpc_name})"
+
+            applies_note = f" (via '*')" if item['applies_via'] == '*' else ""
+
+            Decore.title(f"{rule_name}{applies_note}", width=70)
+            print(f"{'  action':<13}: {action_str}")
+            print(f"{'  operations':<13}: {access_ops}")
+            if target:
+                print(f"{'  target':<13}: {target}")
+            if comment:
+                print(f"{'  comment':<13}: {comment}")
+            print()
+
+
+def show_nacm_user(json):
+    """Display detailed information about a specific user's NACM permissions."""
+    user_name = json.get('_user_name')
+    nacm = json.get("ietf-netconf-acm:nacm", {})
+    system = json.get("ietf-system:system", {})
+
+    # Find the user
+    users = system.get("authentication", {}).get("user", [])
+    user = None
+    for u in users:
+        if u.get("name") == user_name:
+            user = u
+            break
+
+    if not user:
+        print(f"User '{user_name}' not found.")
+        return
+
+    # User details
+    shell_data = user.get("infix-system:shell", "false")
+    shell = shell_data.split(":")[-1] if ":" in shell_data else shell_data
+    print(f"{'shell':<17}: {shell}")
+
+    has_password = bool(user.get("password"))
+    has_keys = bool(user.get("authorized-key"))
+    if has_password and has_keys:
+        login = "password + SSH key"
+    elif has_password:
+        login = "password"
+    elif has_keys:
+        login = "SSH key"
+    else:
+        login = "(none configured)"
+    print(f"{'login':<17}: {login}")
+
+    # Find which NACM groups this user belongs to
+    if not nacm:
+        print("NACM not configured.")
+        return
+
+    groups = nacm.get("groups", {}).get("group", [])
+    user_groups = []
+    for group in groups:
+        if user_name in group.get("user-name", []):
+            user_groups.append(group.get("name"))
+
+    groups_str = ", ".join(user_groups) if user_groups else "(none)"
+    print(f"{'nacm group':<17}: {groups_str}")
+
+    # Show effective permissions for each group
+    if user_groups:
+        permissions = _analyze_group_permissions(nacm)
+
+        for group_name in user_groups:
+            for p in permissions:
+                if p['group'] == group_name:
+                    def perm_str(has_access, has_restrictions):
+                        if not has_access:
+                            return Decore.red("no")
+                        elif has_restrictions:
+                            return Decore.yellow("restricted")
+                        else:
+                            return Decore.green("yes")
+
+                    has_restrictions = bool(p['restrictions'])
+                    print(f"{'read permission':<17}: {perm_str(p['read'], has_restrictions)}")
+                    print(f"{'write permission':<17}: {perm_str(p['write'], has_restrictions)}")
+                    print(f"{'exec permission':<17}: {perm_str(p['exec'], has_restrictions)}")
+                    break
+
+    print()
+    print("For detailed rules, use: show nacm group <name>")
 
 
 def show_system(json):
@@ -4958,6 +5366,8 @@ def main():
               .add_argument('limit', nargs='?', help='Last N lines, default: all')
 
     subparsers.add_parser('show-nacm', help='Show NACM status and groups')
+    subparsers.add_parser('show-nacm-group', help='Show NACM group details')
+    subparsers.add_parser('show-nacm-user', help='Show NACM user details')
 
     subparsers.add_parser('show-ntp', help='Show NTP status') \
               .add_argument('-a', '--address', help='Show details for specific address')
@@ -5024,6 +5434,10 @@ def main():
         show_firewall_logs(args.limit)
     elif args.command == "show-nacm":
         show_nacm(json_data)
+    elif args.command == "show-nacm-group":
+        show_nacm_group(json_data)
+    elif args.command == "show-nacm-user":
+        show_nacm_user(json_data)
     elif args.command == "show-ntp":
         show_ntp(json_data, args.address)
     elif args.command == "show-ntp-tracking":