Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 57 additions & 0 deletions src/dippy/cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,63 @@ def get_description(tokens: list[str], handler_name: str = None) -> str:
return " ".join(tokens[:depth])


def strip_global_flags(tokens: list[str]) -> list[str] | None:
"""Strip handler-recognized global flags from command tokens.

Used by the analyzer to retry config matching when raw tokens don't match.
Looks up the handler for tokens[0] and uses GLOBAL_FLAGS_WITH_ARG and
GLOBAL_FLAGS_NO_ARG constants to remove global flags.

Returns cleaned tokens, or None if no handler, no flags stripped, or
tokens is too short.
"""
if len(tokens) < 2:
return None

handler = get_handler(tokens[0])
if handler is None:
return None

flags_with_arg: frozenset[str] = getattr(
handler, "GLOBAL_FLAGS_WITH_ARG", frozenset()
)
flags_no_arg: frozenset[str] = getattr(handler, "GLOBAL_FLAGS_NO_ARG", frozenset())

if not flags_with_arg and not flags_no_arg:
return None

result = [tokens[0]]
i = 1
changed = False

while i < len(tokens):
token = tokens[i]

# --flag=value form for flags with arg
if any(token.startswith(f"{flag}=") for flag in flags_with_arg):
changed = True
i += 1
continue

# Flags with argument (consume flag + next token)
if token in flags_with_arg:
changed = True
i += 2
continue

# Flags without argument
if token in flags_no_arg:
changed = True
i += 1
continue

# Not a global flag — keep rest as-is
result.extend(tokens[i:])
break

return result if changed else None


def _discover_handlers() -> dict[str, str]:
"""Discover handler modules and build command -> module mapping."""
handlers = {}
Expand Down
36 changes: 26 additions & 10 deletions src/dippy/core/analyzer.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@

from dippy.core.config import Config, match_redirect
from dippy.core.allowlists import SIMPLE_SAFE, WRAPPER_COMMANDS
from dippy.cli import get_handler, get_description, HandlerContext
from dippy.cli import get_handler, get_description, strip_global_flags, HandlerContext
from dippy.vendor.parable import parse, ParseError

# Redirect targets that are always safe (no file write)
Expand Down Expand Up @@ -381,6 +381,16 @@ def _analyze_redirects(
return decisions


def _config_match_decision(match, base: str) -> Decision | None:
"""Convert a config Match to a Decision, or None if no match."""
if not match:
return None
if match.decision == "allow":
return Decision("allow", f"{base} ({match.pattern})")
msg = match.message or match.pattern
return Decision(match.decision, f"{base}: {msg}")


def _analyze_simple_command(
words: list[str], config: Config, cwd: Path, *, remote: bool = False
) -> Decision:
Expand All @@ -404,15 +414,21 @@ def _analyze_simple_command(

cmd = SimpleCommand(words=words)
config_match = match_command(cmd, config, cwd, remote=remote)
if config_match:
if config_match.decision == "allow":
return Decision("allow", f"{base} ({config_match.pattern})")
elif config_match.decision == "deny":
msg = config_match.message or config_match.pattern
return Decision("deny", f"{base}: {msg}")
else: # ask
msg = config_match.message or config_match.pattern
return Decision("ask", f"{base}: {msg}")
decision = _config_match_decision(config_match, base)
if decision:
return decision

# 1.5. If no config match on raw tokens, retry with global flags stripped.
# This lets "allow git commit" match "git -C /path commit".
# Step 1 already handles explicit rules like "deny git -C * commit".
stripped = strip_global_flags(tokens)
if stripped is not None:
stripped_words = list(words[:i]) + stripped if i > 0 else stripped
stripped_cmd = SimpleCommand(words=stripped_words)
config_match = match_command(stripped_cmd, config, cwd, remote=remote)
decision = _config_match_decision(config_match, base)
if decision:
return decision

# 2. Handle wrapper commands (time, timeout, etc.) - analyze inner command
if base in WRAPPER_COMMANDS and len(tokens) > 1:
Expand Down
73 changes: 73 additions & 0 deletions tests/test_analyzer_bugs.py
Original file line number Diff line number Diff line change
Expand Up @@ -522,3 +522,76 @@ def test_cd_tilde_path(self):
config = parse_config(f"allow {home}/script *")
result = analyze("cd ~ && ./script arg", config, Path("/somewhere/else"))
assert result.action == "allow"


class TestConfigGlobalFlagStripping:
"""Config rules should match commands with global flags stripped.

Bug: 'allow git commit' doesn't match 'git -C /path commit -m test'
because _match_words() matches raw tokens and -C /path breaks prefix match.

Fix: After step 1 (raw match) finds no match, strip global flags using
handler constants and re-match (step 1.5).
"""

@pytest.fixture
def cwd(self):
return Path.cwd()

def test_allow_git_commit_with_C_flag(self, cwd):
from dippy.core.config import parse_config

config = parse_config("allow git commit")
result = analyze("git -C /path commit -m test", config, cwd)
assert result.action == "allow"

def test_allow_git_commit_with_work_tree(self, cwd):
from dippy.core.config import parse_config

config = parse_config("allow git commit")
result = analyze("git --work-tree /path commit -m test", config, cwd)
assert result.action == "allow"

def test_allow_git_commit_plain_regression(self, cwd):
from dippy.core.config import parse_config

config = parse_config("allow git commit")
result = analyze("git commit -m test", config, cwd)
assert result.action == "allow"

def test_explicit_deny_with_C_flag_takes_precedence(self, cwd):
"""Step 1 matches deny on raw tokens, step 1.5 never runs."""
from dippy.core.config import parse_config

config = parse_config("allow git commit\ndeny git -C * commit")
result = analyze("git -C /path commit", config, cwd)
assert result.action == "deny"

def test_allow_docker_ps_with_H_flag(self, cwd):
from dippy.core.config import parse_config

config = parse_config("allow docker ps")
result = analyze("docker -H tcp://host ps", config, cwd)
assert result.action == "allow"

def test_allow_git_status_with_no_pager(self, cwd):
from dippy.core.config import parse_config

config = parse_config("allow git status")
result = analyze("git --no-pager status", config, cwd)
assert result.action == "allow"

def test_no_handler_no_stripping(self, cwd):
from dippy.core.config import parse_config

config = parse_config("allow mycmd foo")
result = analyze("mycmd -x /path foo", config, cwd)
assert result.action == "ask"

def test_deny_git_push_with_C_flag(self, cwd):
"""deny rules also work via step 1.5."""
from dippy.core.config import parse_config

config = parse_config("deny git push")
result = analyze("git -C /path push", config, cwd)
assert result.action == "deny"
94 changes: 94 additions & 0 deletions tests/test_strip_global_flags.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
"""Tests for strip_global_flags() in dippy.cli."""

from __future__ import annotations

from dippy.cli import strip_global_flags


class TestStripGlobalFlags:
"""Unit tests for strip_global_flags()."""

# --- Example 1: Git handler ---

def test_git_dash_C_with_arg(self):
result = strip_global_flags(["git", "-C", "/path", "commit", "-m", "test"])
assert result == ["git", "commit", "-m", "test"]

def test_git_git_dir_equals(self):
result = strip_global_flags(["git", "--git-dir=/path", "commit"])
assert result == ["git", "commit"]

def test_git_work_tree_equals(self):
result = strip_global_flags(["git", "--work-tree=/path", "status"])
assert result == ["git", "status"]

def test_git_no_pager_and_dash_C(self):
result = strip_global_flags(["git", "--no-pager", "-C", "/path", "status"])
assert result == ["git", "status"]

def test_git_work_tree_with_arg(self):
result = strip_global_flags(
["git", "--work-tree", "/path", "commit", "-m", "test"]
)
assert result == ["git", "commit", "-m", "test"]

def test_git_multiple_global_flags(self):
result = strip_global_flags(
[
"git",
"--no-pager",
"-c",
"core.editor=vim",
"-C",
"/path",
"log",
"--oneline",
]
)
assert result == ["git", "log", "--oneline"]

def test_git_only_no_arg_flag(self):
result = strip_global_flags(["git", "--no-pager", "status"])
assert result == ["git", "status"]

def test_git_no_flags_returns_none(self):
result = strip_global_flags(["git", "commit", "-m", "test"])
assert result is None

def test_git_dash_c_key_value(self):
result = strip_global_flags(["git", "-c", "user.name=Test", "log"])
assert result == ["git", "log"]

# --- Example 2: Docker handler ---

def test_docker_host_flag(self):
result = strip_global_flags(["docker", "-H", "tcp://host", "ps"])
assert result == ["docker", "ps"]

def test_docker_context_flag(self):
result = strip_global_flags(["docker", "--context", "myctx", "ps"])
assert result == ["docker", "ps"]

def test_docker_no_flags_returns_none(self):
result = strip_global_flags(["docker", "ps"])
assert result is None

# --- No handler / unknown command ---

def test_unknown_command_returns_none(self):
result = strip_global_flags(["unknown-cmd", "-C", "/path", "foo"])
assert result is None

# --- Edge cases ---

def test_empty_tokens(self):
result = strip_global_flags([])
assert result is None

def test_single_token(self):
result = strip_global_flags(["git"])
assert result is None

def test_all_global_flags_no_subcommand(self):
result = strip_global_flags(["git", "--no-pager", "-C", "/path"])
assert result == ["git"]