Skip to content

Fix watcher bugs#17

Open
tastelikefeet wants to merge 2 commits into
modelscope:mainfrom
tastelikefeet:feat/0618
Open

Fix watcher bugs#17
tastelikefeet wants to merge 2 commits into
modelscope:mainfrom
tastelikefeet:feat/0618

Conversation

@tastelikefeet

Copy link
Copy Markdown
Collaborator

PR type

  • Bug Fix
  • New Feature
  • Document Updates
  • Server Enhancement

PR information

Write the detail information belongs to this PR.

Test results

Paste your test result here (if needed).

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request updates the daemon-stopping logic in ultron/cli/watcher.py to support custom pgrep patterns via an optional extra_patterns parameter. The review feedback recommends using Optional[List[str]] instead of string-quoted union types for consistency, deduplicating patterns to avoid redundant pgrep executions, and adding -- to the pgrep command to prevent potential option injection vulnerabilities.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

Comment thread ultron/cli/watcher.py
]


def stop_daemon(extra_patterns: "List[str] | None" = None) -> bool:

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Since Optional is already imported and used elsewhere in this file (e.g., for _logger), we should use Optional[List[str]] instead of the string-quoted "List[str] | None" union syntax for consistency and compatibility.

Suggested change
def stop_daemon(extra_patterns: "List[str] | None" = None) -> bool:
def stop_daemon(extra_patterns: Optional[List[str]] = None) -> bool:

Comment thread ultron/cli/watcher.py
except (OSError, subprocess.TimeoutExpired, ValueError):
pass
return []
def _find_watch_pids(extra_patterns: "List[str] | None" = None) -> List[int]:

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

For consistency with the rest of the file and the use of Optional from typing, use Optional[List[str]] instead of "List[str] | None".

Suggested change
def _find_watch_pids(extra_patterns: "List[str] | None" = None) -> List[int]:
def _find_watch_pids(extra_patterns: Optional[List[str]] = None) -> List[int]:

Comment thread ultron/cli/watcher.py
Comment on lines +289 to +291
patterns = list(_DEFAULT_WATCH_PATTERNS)
if extra_patterns:
patterns.extend(extra_patterns)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

If extra_patterns contains patterns that are already in _DEFAULT_WATCH_PATTERNS or contains duplicate entries, pgrep will be executed multiple times for the same pattern. We can deduplicate the patterns while preserving their order using dict.fromkeys.

Suggested change
patterns = list(_DEFAULT_WATCH_PATTERNS)
if extra_patterns:
patterns.extend(extra_patterns)
patterns = list(dict.fromkeys(_DEFAULT_WATCH_PATTERNS + (extra_patterns or [])))

Comment thread ultron/cli/watcher.py
Comment on lines +295 to +298
result = subprocess.run(
["pgrep", "-f", pattern],
capture_output=True, text=True, timeout=5,
)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

security-medium medium

If any pattern in extra_patterns starts with a hyphen (e.g., -v), it could be interpreted as a command-line option by pgrep, leading to unexpected behavior or option injection. To prevent this, use -- to explicitly signal the end of command options to pgrep.

Suggested change
result = subprocess.run(
["pgrep", "-f", pattern],
capture_output=True, text=True, timeout=5,
)
result = subprocess.run(
["pgrep", "-f", "--", pattern],
capture_output=True, text=True, timeout=5,
)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant