github · voicemodeHEAD

Evidence

Dictionary mapping episode numbers to MfpEpisode objects.

        """

        episodes = {}

        root = ET.fromstring(rss_content)

        for item in root.findall(".//item"):

            enclosure = item.find("enclosure")

Remediation

Use defusedxml in Python (drop-in replacement for stdlib XML modules). In Java, set XMLConstants.FEATURE_SECURE_PROCESSING=true and disable external-DTD features on the factory before parsing.

Evidence

"""Conversation tools for interactive voice interactions."""

import asyncio

import logging

import os

import time

import traceback

from typing import Optional, Literal, Tuple, Dict, Union

from pathlib import Path

from datetime import datetime

import numpy as np

import sounddevice as sd

from scipy.io.wavfile import write

from pydub import AudioSegment

from openai import AsyncOpenAI

import httpx

# Optional webrtcvad for silence detection

try:

    import webrtcvad

    VAD_AVAILABLE = True

except 

Remediation

Pick one confirmation mechanism and add it to the destructive handler: - make the client confirm via `elicit()` before touching the sink, - accept a `dry_run` / `dryRun` boolean input and branch off the sink when true (return what would change), - require an `idempotency_key` / `idempotencyKey` the caller provides on the first call and echoes back to commit, - require a `confirmation` / `confirm` token as an input field. If the destructive call is genuinely irrelevant to the tool

Evidence

"""Uninstall tool for Whisper STT service."""

import os

import shutil

import subprocess

import platform

import logging

from pathlib import Path

from typing import Dict, Any, Union

from voice_mode.server import mcp

from voice_mode.config import BASE_DIR

from voice_mode.utils.services.common import find_process_by_port

logger = logging.getLogger("voicemode")

@mcp.tool()

async def whisper_uninstall(

    remove_models: Union[bool, str] = False,

    remove_all_data: Union[bool, str] = False

) ->

Remediation

Pick one confirmation mechanism and add it to the destructive handler: - make the client confirm via `elicit()` before touching the sink, - accept a `dry_run` / `dryRun` boolean input and branch off the sink when true (return what would change), - require an `idempotency_key` / `idempotencyKey` the caller provides on the first call and echoes back to commit, - require a `confirmation` / `confirm` token as an input field. If the destructive call is genuinely irrelevant to the tool

Evidence

"""Uninstall tool for Kokoro TTS service."""

import os

import shutil

import subprocess

import platform

import logging

from pathlib import Path

from typing import Dict, Any, Union

from voice_mode.server import mcp

from voice_mode.config import BASE_DIR

from voice_mode.utils.services.common import find_process_by_port

logger = logging.getLogger("voicemode")

@mcp.tool()

async def kokoro_uninstall(

    remove_models: Union[bool, str] = False,

    remove_all_data: Union[bool, str] = False

) -> D

Remediation

Pick one confirmation mechanism and add it to the destructive handler: - make the client confirm via `elicit()` before touching the sink, - accept a `dry_run` / `dryRun` boolean input and branch off the sink when true (return what would change), - require an `idempotency_key` / `idempotencyKey` the caller provides on the first call and echoes back to commit, - require a `confirmation` / `confirm` token as an input field. If the destructive call is genuinely irrelevant to the tool

Evidence

"""Installation tool for kokoro-fastapi TTS service."""

import os

import sys

import platform

import subprocess

import shutil

import logging

from pathlib import Path

from typing import Dict, Any, Optional, Union

import asyncio

import aiohttp

from voice_mode.server import mcp

from voice_mode.config import SERVICE_AUTO_ENABLE

from voice_mode.utils.version_helpers import (

    get_git_tags, get_latest_stable_tag, get_current_version,

    checkout_version, is_version_installed

)

from voice_mode.uti

Remediation

Pick one confirmation mechanism and add it to the destructive handler: - make the client confirm via `elicit()` before touching the sink, - accept a `dry_run` / `dryRun` boolean input and branch off the sink when true (return what would change), - require an `idempotency_key` / `idempotencyKey` the caller provides on the first call and echoes back to commit, - require a `confirmation` / `confirm` token as an input field. If the destructive call is genuinely irrelevant to the tool

Evidence

"""Installation tool for whisper.cpp"""

import os

import sys

import platform

import subprocess

import shutil

import json

import logging

from pathlib import Path

from typing import Dict, Any, Optional, Union

import asyncio

import aiohttp

try:

    from importlib.resources import files

except ImportError:

    # Python < 3.9 fallback

    from importlib_resources import files

from voice_mode.server import mcp

from voice_mode.config import SERVICE_AUTO_ENABLE, DEFAULT_WHISPER_MODEL, WHISPER_PORT

fro

Remediation

Pick one confirmation mechanism and add it to the destructive handler: - make the client confirm via `elicit()` before touching the sink, - accept a `dry_run` / `dryRun` boolean input and branch off the sink when true (return what would change), - require an `idempotency_key` / `idempotencyKey` the caller provides on the first call and echoes back to commit, - require a `confirmation` / `confirm` token as an input field. If the destructive call is genuinely irrelevant to the tool

Evidence

"""Unified service management tool for voice mode services."""

import asyncio

import logging

import os

import platform

import subprocess

import time

from pathlib import Path

from typing import Literal, Optional, Dict, Any, Union

import psutil

from voice_mode.server import mcp

from voice_mode.config import WHISPER_PORT, KOKORO_PORT, MLX_AUDIO_PORT, SERVICE_AUTO_ENABLE

from voice_mode.utils.services.common import find_process_by_port, check_service_status

# Default port for VoiceMode serve com

Remediation

Pick one confirmation mechanism and add it to the destructive handler: - make the client confirm via `elicit()` before touching the sink, - accept a `dry_run` / `dryRun` boolean input and branch off the sink when true (return what would change), - require an `idempotency_key` / `idempotencyKey` the caller provides on the first call and echoes back to commit, - require a `confirmation` / `confirm` token as an input field. If the destructive call is genuinely irrelevant to the tool

Evidence

success = event['data'].get('success', True)

            details = f"tool={tool}, success={success}"

        print(f"{elapsed:>7.3f}s {event['event_type']:25} {details}")

def calculate_ai_thinking_times(all_events: List[Dict[str, Any]]) -> List[float]:

Remediation

Strip C0/C1 control codes before printing user-controlled values. Python: re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", s). Prefer a structured logger (json/logfmt) over raw print to stdout.

Evidence

if __name__ == "__main__":

    if len(sys.argv) != 3:

        print("Usage: process-readme-for-docs.py <input> <output>")

        sys.exit(1)

    process_readme(sys.argv[1], sys.argv[2])

Remediation

Strip C0/C1 control codes before printing user-controlled values. Python: re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", s). Prefer a structured logger (json/logfmt) over raw print to stdout.

Evidence

event = data.decode('utf-8', errors='ignore')

                    if event.endswith('_1'):  # Touch needed

                        print(f"YubiKey touch needed: {event}")

                        play_notification()

        except Exception as e:

Remediation

Strip C0/C1 control codes before printing user-controlled values. Python: re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", s). Prefer a structured logger (json/logfmt) over raw print to stdout.

Evidence

result = subprocess.run(cmd, shell=shell, capture_output=True, text=True)

        return result.returncode == 0, result.stdout, result.stderr

    except Exception as e:

        return False, "", str(e)

def check_wsl_version():

    """Check if running in WSL and get version info"""

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

except Exception as e:

        logger.error(f"Error viewing logs for {service_name}: {e}")

        return f"❌ Error viewing logs: {str(e)}"

@mcp.tool()

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

except Exception as e:

        logger.error(f"Error stopping {service_name}: {e}")

        return f"❌ Error stopping {service_name}: {str(e)}"

async def restart_service(service_name: str) -> str:

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

except Exception as e:

        logger.error(f"Error showing recent statistics: {e}")

        return f"Error showing recent statistics: {str(e)}"

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

try:

                return path.read_text()

            except Exception as e:

                return f"Error reading CHANGELOG.md from {path}: {str(e)}"

    return """CHANGELOG.md not found in package.

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

except Exception as e:

        logger.error(f"Error generating statistics summary: {e}")

        return f"Error generating statistics summary: {str(e)}"

@mcp.tool()

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

info["running"] = True

            info["pid"] = proc.pid

            try:

                info["uptime_seconds"] = int(proc.create_time())

            except:

                pass

    except Exception as e:

        info["error"] = str(e)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

["launchctl", "remove", service_name],

                                    capture_output=True

                                )

                                removed_items.append(f"Removed launchctl instance: {service_name}")

                            except Exception:

                                pass

        elif system == "Linux":

            # Remove systemd service

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

_play_audio(tmp_path)

        finally:

            try:

                os.unlink(tmp_path)

            except OSError:

                pass

def _sayas_preview(voice: str, profile: dict) -> None:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if not track and track_path.is_absolute():

        try:

            rel_path = str(track_path.relative_to(library.music_root))

            track = library.get_track_by_path(rel_path)

        except ValueError:

            pass

    if not track:

        click.echo(f"Track not found in library: {status.path}", err=True)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

)

        if result.returncode == 0:

            data = json.loads(result.stdout)

            return data['info']['version']

    except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired, json.JSONDecodeError, KeyError):

        pass

    return None

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.