removeaudio-code (Python utility for removing audio tracks from media files)

noaudio.py

#!/usr/bin/env python3"""⚠️ DEPRECATED - DO NOT USE ⚠️This script has critical security vulnerabilities:- Command Injection (CVSS 9.8)- Path Traversal- Poor Error HandlingPlease use noaudio_secure.py instead.This file is kept for reference only."""import sysprint("=" * 70)print("⚠️  SECURITY WARNING ⚠️")print("=" * 70)print()print("This script (noaudio.py) has CRITICAL security vulnerabilities:")print("  - Command Injection (CVSS 9.8)")print("  - Path Traversal")print("  - Poor Error Handling")print()print("Please use noaudio_secure.py instead:")print("  python noaudio_secure.py")print()print("=" * 70)sys.exit(1)

noaudio.py.VULNERABLE

from os import errorimport subprocessimport osPOSSIBLE_EXTENSIONS = ["avi", "mp4", "mov"]#MAKE SURE DIRECTORY DOES NOT HAVE SPACESdirectory = input("Give directory path containing video footage: ")directory = directory.replace("\\", "/")print(f"Searching in {directory}")if os.path.exists(directory):    try:        for files in os.walk(directory):            for l in files:                for filename in l:                    if '.' in filename:                        ext = filename.split('.')[1]                        print(ext)                        if ext.lower() in POSSIBLE_EXTENSIONS:                            noaudio = "noaudio_" + filename                            oldpath = os.path.join(directory, filename)                            newpath = path = os.path.join(directory, noaudio)                            command = 'ffmpeg -i ' + oldpath + ' -c copy -an ' + newpath                            subprocess.call(command, shell=True)                            print('Removed audio')        print('Done, exiting')    except Exception as e:        print(error, e)        print('Error, exiting')else:    print("No video footage found, try again with another directory path.")

noaudio_secure.py

#!/usr/bin/env python3"""Secure video audio removal toolFixes: Command injection, path traversal, error handling"""import subprocessimport osimport sysfrom pathlib import Pathimport tempfilePOSSIBLE_EXTENSIONS = ["avi", "mp4", "mov"]def validate_directory(directory_path):    """Validate and sanitize directory path"""    try:        # Convert to absolute path and resolve symlinks        path = Path(directory_path).resolve()                # Check if directory exists        if not path.exists():            print(f"❌ Error: Directory does not exist: {path}")            return None                # Check if it's actually a directory        if not path.is_dir():            print(f"❌ Error: Path is not a directory: {path}")            return None                # Check read permissions        if not os.access(path, os.R_OK):            print(f"❌ Error: No read permission for directory: {path}")            return None                return path    except Exception as e:        print(f"❌ Error validating directory: {e}")        return Nonedef check_ffmpeg():    """Check if ffmpeg is installed and accessible"""    try:        result = subprocess.run(            ["ffmpeg", "-version"],            capture_output=True,            text=True,            timeout=5        )        if result.returncode == 0:            return True        else:            print("❌ Error: ffmpeg is installed but not working correctly")            return False    except FileNotFoundError:        print("❌ Error: ffmpeg is not installed or not in PATH")        print("Please install ffmpeg:")        print("  Windows: choco install ffmpeg")        print("  macOS: brew install ffmpeg")        print("  Linux: sudo apt-get install ffmpeg")        return False    except Exception as e:        print(f"❌ Error checking ffmpeg: {e}")        return Falsedef remove_audio_from_video(input_path, output_path):    """    Remove audio from video file using ffmpeg    Uses secure subprocess call with list arguments    """    temporary_path = None    try:        input_path = Path(input_path)        output_path = Path(output_path)        if not input_path.is_file():            return False, "Input is not a readable video file"        if output_path.exists() or output_path.is_symlink():            return False, "Output already exists; it was preserved"        # Stage beside the destination so publication stays on one filesystem.        descriptor, name = tempfile.mkstemp(            prefix='.noaudio-', suffix=output_path.suffix, dir=output_path.parent        )        os.close(descriptor)        temporary_path = Path(name)        # Build command as list (prevents command injection)        command = [            "ffmpeg",            "-nostdin",            "-i", str(input_path),            "-c", "copy",            "-an",            str(temporary_path),            "-y"  # Only overwrite this invocation's newly created staging file        ]                # Execute with shell=False (secure)        result = subprocess.run(            command,            capture_output=True,            text=True,            encoding="utf-8",            errors="replace",            timeout=300  # 5 minute timeout        )                if result.returncode == 0:            if not temporary_path.is_file() or temporary_path.stat().st_size == 0:                return False, "FFmpeg did not produce a non-empty video"            # Both operations fail if another writer created the destination.            # POSIX rename would overwrite it, so use an atomic hard link there.            if os.name == 'nt':                os.rename(temporary_path, output_path)            else:                os.link(temporary_path, output_path)            return True, "Success"        else:            return False, f"ffmpeg error: {result.stderr}"        except subprocess.TimeoutExpired:        return False, "Timeout: Video processing took too long"    except Exception as e:        return False, f"Error: {str(e)}"    finally:        if temporary_path is not None:            try:                temporary_path.unlink(missing_ok=True)            except OSError:                print(f"⚠️ Could not remove temporary output: {temporary_path}")def process_directory(directory):    """Process all video files in directory"""    processed = 0    failed = 0    skipped = 0        print(f"🔍 Searching for video files in: {directory}")    print(f"📹 Looking for extensions: {', '.join(POSSIBLE_EXTENSIONS)}")    print()        # Walk through directory    for root, dirs, files in os.walk(directory):        for filename in files:            # Check if file has valid extension            if '.' not in filename:                continue                        ext = filename.rsplit('.', 1)[-1].lower()            if ext not in POSSIBLE_EXTENSIONS:                continue                        # Skip files that already have "noaudio_" prefix            if filename.startswith(("noaudio_", ".noaudio-")):                skipped += 1                continue                        # Build paths            input_path = Path(root) / filename            output_filename = f"noaudio_{filename}"            output_path = Path(root) / output_filename                        # Check if output file already exists            if output_path.exists():                print(f"⏭️  Skipping (output exists): {filename}")                skipped += 1                continue                        print(f"🎬 Processing: {filename}")                        # Remove audio            success, message = remove_audio_from_video(input_path, output_path)                        if success:                print(f"✅ Success: {output_filename}")                processed += 1            else:                print(f"❌ Failed: {filename} - {message}")                failed += 1                        print()        # Summary    print("=" * 60)    print("📊 PROCESSING SUMMARY")    print("=" * 60)    print(f"✅ Processed successfully: {processed}")    print(f"❌ Failed: {failed}")    print(f"⏭️  Skipped: {skipped}")    print(f"📁 Total files: {processed + failed + skipped}")    print()    return {"processed": processed, "failed": failed, "skipped": skipped}def main():    """Main execution flow"""    print("=" * 60)    print("🎬 VIDEO AUDIO REMOVAL TOOL")    print("=" * 60)    print()        # Check ffmpeg    print("🔧 Checking dependencies...")    if not check_ffmpeg():        sys.exit(1)    print("✅ ffmpeg is installed and working")    print()        # Get directory from user    directory_input = input("📁 Enter directory path containing video files: ").strip()        # Validate directory    directory = validate_directory(directory_input)    if directory is None:        sys.exit(1)        print(f"✅ Directory validated: {directory}")    print()        # Confirm before processing    response = input("⚠️  This will create new files with 'noaudio_' prefix. Continue? (yes/no): ")    if response.lower() != "yes":        print("❌ Operation cancelled by user")        sys.exit(0)        print()        # Process directory    try:        counts = process_directory(directory)        if counts["failed"]:            print("❌ Some videos failed; completed outputs were preserved.")            sys.exit(1)        print("✅ Done!")    except KeyboardInterrupt:        print("\n\n⚠️  Operation interrupted by user")        sys.exit(1)    except Exception as e:        print(f"\n\n❌ Unexpected error: {e}")        sys.exit(1)if __name__ == "__main__":    main()

STANDARDS.md

Repository StandardsThis document is the authoritative list of standards every repository in thisworkspace is expected to follow. New repositories created fromtheprawntemplate inherit the baseline automatically; this file records whatthat baseline is, what is required vs. optional, and the packaging pattern forcontainer repositories.It exists so that humans and automation agents working across the fleet have asingle source of truth to conform to, rather than relying on implicitper-repository conventions.Last reviewed: 2026-09-27.## ScopeApplies to all owned repositories (currently 79: 71 non-container, 7 container,plus one history-managed repository). "Container repository" means a repositorythat builds and publishes one or more container images to GitHub ContainerRegistry (GHCR, ghcr.io).## 1. Security scanning (REQUIRED for every repo)Every repository must carry these GitHub Actions workflows. They are present intheprawntemplate and propagate to new repositories:| Workflow | File | Purpose || --- | --- | --- || CodeQL | .github/workflows/codeql.yml | Static analysis / code scanning || Semgrep | .github/workflows/semgrep.yml | Pattern-based SAST || TruffleHog | .github/workflows/trufflehog.yml | Secret scanning || Dependency Review | .github/workflows/dependency-review.yml | Blocks vulnerable dependency changes in PRs || Scorecard | .github/workflows/scorecard.yml | OpenSSF supply-chain posture || LFS Guard | .github/workflows/lfs-guard.yml | Blocks accidental large/binary commits |A repository is security-compliant when all six are present and enabled.## 2. Automation and CI (REQUIRED)- .github/workflows/ci.yml ΓÇö project checks; adapt commands to the stack.- .github/dependabot.yml + .github/workflows/dependabot-auto-merge.yml and .github/workflows/auto-merge-bots.yml ΓÇö dependency updates with guarded auto-merge.- .github/workflows/labeler.yml, greetings.yml, heartbeat.yml, summary.yml ΓÇö repository hygiene and activity signals.- .github/workflows/deploy.yml ΓÇö deployment (or a clear statement that the repo is not deployed).## 3. Governance and hygiene files (REQUIRED unless noted)| File | Requirement || --- | --- || LICENSE | Required ΓÇö Apache-2.0. || NOTICE | Required ΓÇö accompanies Apache-2.0. || SECURITY.md | Required. Use GitHub private "Report a vulnerability"; never a personal email. || CONTRIBUTING.md | Required. || AGENTS.md | Required ΓÇö instructions for automation agents. || .github/dependabot.yml | Required. || .env.example | Required only when the project reads environment variables. Repos with no env config legitimately omit it. || .github/ISSUE_TEMPLATE/*, .github/pull_request_template.md | Required. || .gitignore, .gitattributes | Required. |## 4. Privacy and portability rules (REQUIRED)These reflect the fleet-wide remediation sweep and must hold for every repo:- No secrets in git. Use environment variables and .env.example. TruffleHog enforces this in CI; do not defeat it.- No personal contact details or private operator references in tracked files (READMEs, SECURITY.md, configs, .agents/). Vulnerability reporting goes through GitHub's private channel, not a personal email.- Windows/Linux portability. Launchers and paths must work on both; avoid hardcoded absolute paths and OS-specific-only launch scripts where a portable equivalent exists.- .agents/ hygiene. Never write secrets or personal details into .agents/STATE.md or .agents/JOURNAL.md.- PRs over direct commits to main. Changes land through reviewed pull requests.## 5. Container / GHCR packaging (REQUIRED for container repos; OPT-IN for others)Container repositories publish images to ghcr.io/<owner>/<repo>/<component>.The standard packaging pattern ΓÇö verified present in all seven current containerrepositories (musicstream, facetracker, FORGE, unifiedcollector, unifiedanalyzer,ticketremaster-b, ctfsolver) ΓÇö is:| File | Purpose || --- | --- || .github/workflows/docker-publish.yml | Builds and pushes image(s) to GHCR on the release trigger. || .github/scripts/ghcr-retention.cjs | Prunes old GHCR package versions to control storage. || Dockerfile (and variants) | Image build definition(s). |Opt-in scaffolding for these files lives under.github/optional/ so a new repository can adopt thepackaging pattern by copying them into place rather than hand-rolling each time.See .github/optional/README.md for enable steps.Package visibility: package visibility (private vs public) is governed by theorganization's package-creation policy (Org Settings ΓåÆ Packages). Individualcontainer package visibility is a deliberate, per-package decision ΓÇö especiallyfor privacy-sensitive images (photo/OSINT/security tooling) whose layers becomeworld-pullable when public.## 6. Compliance statusA live compliance matrix across all owned repositories is maintained out-of-band(see repo-standards-compliance-matrix.md in the workspace release artifacts).As of the last review, 76/79 repositories carry the full security-workflow set;the known exceptions are actively-developed repositories that have not yetadopted the baseline. All seven container repositories carry the complete GHCRpipeline.## How to use this document- Creating a new repo: start from theprawntemplate; you inherit sections 1ΓÇô4 automatically. If it will publish containers, adopt section 5 from .github/optional/.- Auditing an existing repo: check it against sections 1ΓÇô4 (all repos) and section 5 (container repos). Missing required items are compliance gaps to fix via a reviewed PR.- Agents: treat this file as the conformance target for any repository work.

tests__test_noaudio.py

"""Exercise conversion failures with synthetic files and a fake FFmpeg process."""import contextlibimport iofrom pathlib import Pathimport subprocessfrom tempfile import TemporaryDirectoryimport unittestfrom unittest.mock import patchimport noaudio_secure as audioclass AudioOutputTests(unittest.TestCase):    def setUp(self):        self.temporary = TemporaryDirectory()        self.addCleanup(self.temporary.cleanup)        self.root = Path(self.temporary.name)        self.source = self.root / 'synthetic input.mp4'        self.source.write_bytes(b'synthetic original video')        self.output = self.root / 'noaudio_synthetic input.mp4'        self.console = io.StringIO()    def convert(self, fake):        with patch.object(audio.subprocess, 'run', side_effect=fake) as process:            result = audio.remove_audio_from_video(self.source, self.output)        self.assertEqual(self.source.read_bytes(), b'synthetic original video')        return result, process    def test_success_publishes_finished_file_and_cleans_staging(self):        def success(command, **kwargs):            self.assertEqual(kwargs['timeout'], 300)            self.assertIn('-nostdin', command)            target = Path(command[-2])            self.assertNotEqual(target, self.output)            self.assertFalse(self.output.exists())            target.write_bytes(b'complete muted video')            return subprocess.CompletedProcess(command, 0, '', '')        result, _ = self.convert(success)        self.assertEqual(result, (True, 'Success'))        self.assertEqual(self.output.read_bytes(), b'complete muted video')        self.assertEqual(set(self.root.iterdir()), {self.source, self.output})    def test_failed_conversion_leaves_no_completed_output(self):        def failure(command, **kwargs):            Path(command[-2]).write_bytes(b'incomplete video')            return subprocess.CompletedProcess(command, 1, '', 'synthetic conversion error')        result, _ = self.convert(failure)        self.assertFalse(result[0])        self.assertFalse(self.output.exists())        self.assertEqual(list(self.root.iterdir()), [self.source])    def test_timeout_removes_only_its_temporary_file(self):        def timeout(command, **kwargs):            Path(command[-2]).write_bytes(b'incomplete video')            raise subprocess.TimeoutExpired(command, kwargs['timeout'])        result, _ = self.convert(timeout)        self.assertFalse(result[0])        self.assertFalse(self.output.exists())        self.assertEqual(list(self.root.iterdir()), [self.source])    def test_existing_output_is_preserved_without_starting_ffmpeg(self):        self.output.write_bytes(b'existing user output')        result, process = self.convert(AssertionError('Must not start FFmpeg'))        self.assertFalse(result[0])        process.assert_not_called()        self.assertEqual(self.output.read_bytes(), b'existing user output')    def test_empty_success_is_not_published(self):        result, _ = self.convert(lambda command, **kwargs: subprocess.CompletedProcess(command, 0, '', ''))        self.assertFalse(result[0])        self.assertEqual(list(self.root.iterdir()), [self.source])    def test_interrupt_cleans_staging_and_propagates(self):        def interrupt(command, **kwargs):            Path(command[-2]).write_bytes(b'incomplete video')            raise KeyboardInterrupt        with self.assertRaises(KeyboardInterrupt):            self.convert(interrupt)        self.assertEqual(list(self.root.iterdir()), [self.source])        self.assertEqual(self.source.read_bytes(), b'synthetic original video')    def test_output_created_during_conversion_is_not_overwritten(self):        def race(command, **kwargs):            Path(command[-2]).write_bytes(b'new muted video')            self.output.write_bytes(b'concurrent user output')            return subprocess.CompletedProcess(command, 0, '', '')        result, _ = self.convert(race)        self.assertFalse(result[0])        self.assertEqual(self.output.read_bytes(), b'concurrent user output')        self.assertEqual(set(self.root.iterdir()), {self.source, self.output})    def test_directory_reports_failures_and_preserves_existing_outputs(self):        other = self.root / 'other.MOV'        other.write_bytes(b'synthetic other')        self.output.write_bytes(b'existing output')        (self.root / '.noaudio-interrupted.mp4').write_bytes(b'old incomplete temporary video')        with patch.object(audio, 'remove_audio_from_video', return_value=(False, 'synthetic error')) as convert:            with contextlib.redirect_stdout(self.console):                counts = audio.process_directory(self.root)        self.assertEqual(counts, {'processed': 0, 'failed': 1, 'skipped': 3})        convert.assert_called_once_with(other, self.root / 'noaudio_other.MOV')    def test_cli_reports_failed_batch_with_nonzero_exit(self):        with patch.object(audio, 'check_ffmpeg', return_value=True), \                patch.object(audio, 'validate_directory', return_value=self.root), \                patch.object(audio, 'process_directory', return_value={'processed': 0, 'failed': 1, 'skipped': 0}), \                patch('builtins.input', side_effect=[str(self.root), 'yes']), \                contextlib.redirect_stdout(self.console):            with self.assertRaises(SystemExit) as raised:                audio.main()        self.assertEqual(raised.exception.code, 1)        self.assertNotIn('Done!', self.console.getvalue())if __name__ == '__main__':    unittest.main()