tor

Dockerfile

@@ -6,5 +6,7 @@ RUN apt-get update
 RUN apt-get install -y libopus0 opus-tools ffmpeg
 COPY requirements.txt requirements.txt
 RUN pip install -r requirements.txt
+RUN apt-get install -y tor obfs4proxy
 COPY v6d3music v6d3music
+RUN printf "\nClientTransportPlugin obfs4 exec /usr/bin/obfs4proxy\nBridge obfs4 65.108.56.114:55487 621BA99387F65441630DFBC8A403D11D126EBC72 cert=5HzsLradvYOipNky+aHrgo31KRtxq5Cb6tz3y5Ds7PbBeB0r+C4r15IPYppupCJgzuXgWw iat-mode=0\nUseBridges 1\n" >> "/etc/tor/torrc"
 CMD ["python3", "-m", "v6d3music.run-bot"]

v6d3music/extract.py

@@ -0,0 +1,8 @@
+import youtube_dl
+
+
+def extract(params: dict, url: str, kwargs: dict):
+    extracted = youtube_dl.YoutubeDL(params=params).extract_info(url, **kwargs)
+    if 'entries' in extracted:
+        extracted['entries'] = list(extracted['entries'])
+    return extracted

v6d3music/ffmpegtoraudio.py

@@ -0,0 +1,33 @@
+import shlex
+import subprocess
+
+import discord
+
+
+class FFmpegTorAudio(discord.FFmpegAudio):
+    def __init__(self, source, *, executable='ffmpeg', pipe=False, stderr=None, before_options=None, options=None):
+        args = [executable]
+        subprocess_kwargs = {'stdin': source if pipe else subprocess.DEVNULL, 'stderr': stderr}
+
+        if isinstance(before_options, str):
+            args.extend(shlex.split(before_options))
+
+        args.append('-i')
+        args.append('-' if pipe else source)
+        args.extend(('-f', 's16le', '-ar', '48000', '-ac', '2', '-loglevel', 'warning'))
+
+        if isinstance(options, str):
+            args.extend(shlex.split(options))
+
+        args.append('pipe:1')
+
+        super().__init__(source, executable='torify', args=args, **subprocess_kwargs)
+
+    def read(self):
+        ret = self._stdout.read(discord.opus.Encoder.FRAME_SIZE)
+        if len(ret) != discord.opus.Encoder.FRAME_SIZE:
+            return b''
+        return ret
+
+    def is_opus(self):
+        return False

v6d3music/run-bot.py

@@ -1,5 +1,6 @@
 import asyncio
 import concurrent.futures
+import json
 import os
 import random
 import re
@@ -9,12 +10,11 @@ import subprocess
 import time
 from collections import deque
 from io import StringIO
-from typing import Optional, AsyncIterable, Any, Iterable
+from typing import Optional, AsyncIterable, Any, Iterable, TypeAlias
 
 # noinspection PyPackageRequirements
 import discord
 import nacl.hash
-import youtube_dl
 from ptvp35 import Db, KVJson
 from v6d1tokens.client import request_token
 from v6d2ctx.context import Context, at, escape, monitor, Benchmark, Explicit, Implicit
@@ -22,6 +22,8 @@ from v6d2ctx.handle_content import handle_content
 from v6d2ctx.lock_for import lock_for
 from v6d2ctx.serve import serve
 
+import v6d3music.extract
+import v6d3music.ffmpegtoraudio
 from v6d3music.config import prefix, myroot
 
 loop = asyncio.new_event_loop()
@@ -118,7 +120,7 @@ def sparq(options: str) -> float:
 
 
 class YTAudio(discord.AudioSource):
-    source: discord.FFmpegPCMAudio
+    source: discord.FFmpegAudio
 
     def __init__(
             self,
@@ -128,6 +130,7 @@ class YTAudio(discord.AudioSource):
             options: Optional[str],
             rby: discord.Member,
             already_read: int,
+            tor: bool
     ):
         self.url = url
         self.origin = origin
@@ -135,6 +138,7 @@ class YTAudio(discord.AudioSource):
         self.options = options
         self.rby = rby
         self.already_read = already_read
+        self.tor = tor
         self.loaded = False
         self.regenerating = False
         self.set_source()
@@ -142,7 +146,10 @@ class YTAudio(discord.AudioSource):
 
     def set_source(self):
         self.schedule_duration_update()
-        self.source = discord.FFmpegPCMAudio(
+        audio_class = discord.FFmpegPCMAudio
+        if self.tor:
+            audio_class = v6d3music.ffmpegtoraudio.FFmpegTorAudio
+        self.source = audio_class(
             self.url,
             options=self.options,
             before_options=self.before_options()
@@ -172,9 +179,15 @@ class YTAudio(discord.AudioSource):
         if url in self._durations:
             return
         self._durations.setdefault(url, '')
-        p = subprocess.Popen(
+        prompt = ''
+        if self.tor:
+            prompt = 'torify '
+        prompt += (
             f'ffprobe -i {shlex.quote(url)}'
-            ' -show_entries format=duration -v quiet -of csv="p=0" -sexagesimal',
+            ' -show_entries format=duration -v quiet -of csv="p=0" -sexagesimal'
+        )
+        p = subprocess.Popen(
+            prompt,
             stdout=subprocess.PIPE,
             shell=True
         )
@@ -245,6 +258,7 @@ class YTAudio(discord.AudioSource):
             'options': self.options,
             'rby': self.rby.id,
             'already_read': self.already_read,
+            'tor': self.tor,
         }
 
     @classmethod
@@ -255,13 +269,14 @@ class YTAudio(discord.AudioSource):
             respawn['description'],
             respawn['options'],
             guild.get_member(respawn['rby']) or await guild.fetch_member(respawn['rby']),
-            respawn['already_read']
+            respawn['already_read'],
+            respawn.get('tor', False),
         )
 
     async def regenerate(self):
         try:
             print(f'regenerating {self.origin}')
-            self.url = await real_url(self.origin, True)
+            self.url = await real_url(self.origin, True, self.tor)
             self.source.cleanup()
             self.set_source()
             print(f'regenerated {self.origin}')
@@ -387,13 +402,6 @@ class MainAudio(discord.PCMVolumeTransformer):
         await volume_db.set(member.guild.id, volume)
 
 
-def extract(params: dict, url: str, kwargs: dict):
-    extracted = youtube_dl.YoutubeDL(params=params).extract_info(url, **kwargs)
-    if 'entries' in extracted:
-        extracted['entries'] = list(extracted['entries'])
-    return extracted
-
-
 def bytes_hash(b: bytes) -> str:
     return nacl.hash.sha256(b).decode()
 
@@ -414,14 +422,33 @@ async def aextract(params: dict, url: str, **kwargs):
         with concurrent.futures.ProcessPoolExecutor() as pool:
             return await loop.run_in_executor(
                 pool,
-                extract,
+                v6d3music.extract.extract,
                 params,
                 url,
                 kwargs
             )
 
 
-async def cache_url(hurl: str, rurl: str, override: bool) -> None:
+async def tor_extract(params: dict, url: str, **kwargs):
+    print(f'tor extracting {url}')
+    p = subprocess.Popen(
+        ['torify', 'python', '-m', 'v6d3music.run-extract'],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        text=True
+    )
+    p.stdin.write(f'{json.dumps(params)}\n')
+    p.stdin.write(f'{json.dumps(url)}\n')
+    p.stdin.write(f'{json.dumps(kwargs)}\n')
+    p.stdin.flush()
+    p.stdin.close()
+    code = await loop.run_in_executor(None, p.wait)
+    if code:
+        raise RuntimeError(code)
+    return json.loads(p.stdout.read())
+
+
+async def cache_url(hurl: str, rurl: str, override: bool, tor: bool) -> None:
     async with lock_for(('cache', hurl), 'cache failed'):
         if not override and cache_db.get(f'url:{hurl}', None) is not None:
             return
@@ -429,15 +456,19 @@ async def cache_url(hurl: str, rurl: str, override: bool) -> None:
         if cachable:
             print('caching', hurl)
             path = cache_root / f'{hurl}.opus'
-            p = subprocess.Popen(
+            args = []
+            if tor:
+                args.append('torify')
+            args.extend(
                 [
                     'ffmpeg', '-hide_banner', '-loglevel', 'warning',
                     '-reconnect', '1', '-reconnect_at_eof', '0',
                     '-reconnect_streamed', '1', '-reconnect_delay_max', '10', '-copy_unknown',
                     '-y', '-i', rurl, str(path)
-                ],
+                ]
-                # stdout=subprocess.PIPE,
+            )
-                # stderr=subprocess.PIPE,
+            p = subprocess.Popen(
+                args,
             )
             with Benchmark('CCH'):
                 code = await loop.run_in_executor(None, p.wait)
@@ -450,15 +481,23 @@ async def cache_url(hurl: str, rurl: str, override: bool) -> None:
             await cache_db.set(f'cachable:{hurl}', True)
 
 
-async def real_url(url: str, override: bool) -> str:
+async def real_url(url: str, override: bool, tor: bool) -> str:
     hurl: str = bytes_hash(url.encode())
     if not override:
         curl: Optional[str] = cache_db.get(f'url:{hurl}', None)
         if curl is not None:
             print('using cached', hurl)
             return curl
+    args = []
+    if tor:
+        args.append('torify')
+    args.extend(
+        [
+            'youtube-dl', '--no-playlist', '-f', 'bestaudio', '-g', '--', url
+        ]
+    )
     p = subprocess.Popen(
-        ['youtube-dl', '--no-playlist', '-f', 'bestaudio', '-g', '--', url],
+        args,
         stdout=subprocess.PIPE
     )
     with Benchmark('URL'):
@@ -466,7 +505,7 @@ async def real_url(url: str, override: bool) -> str:
     if code:
         raise RuntimeError(code)
     rurl: str = p.stdout.readline().decode()[:-1]
-    loop.create_task(cache_url(hurl, rurl, override))
+    loop.create_task(cache_url(hurl, rurl, override, tor))
     return rurl
 
 
@@ -474,7 +513,9 @@ def options_for_effects(effects: str) -> Optional[str]:
     return f'-af {shlex.quote(effects)}' if effects else None
 
 
-async def create_ytaudio(ctx: Context, info: dict[str, Any], effects: Optional[str], already_read: int) -> YTAudio:
+async def create_ytaudio(
+        ctx: Context, info: dict[str, Any], effects: Optional[str], already_read: int, tor: bool
+) -> YTAudio:
     if effects:
         if effects not in allowed_effects:
             assert_admin(ctx.member)
@@ -484,19 +525,23 @@ async def create_ytaudio(ctx: Context, info: dict[str, Any], effects: Optional[s
     else:
         options = None
     return YTAudio(
-        await real_url(info['url'], False),
+        await real_url(info['url'], False, tor),
         info['url'],
         f'{escape(info.get("title", "unknown"))} `Rby` {ctx.member}',
         options,
         ctx.member,
-        already_read
+        already_read,
+        tor
     )
 
 
-async def entries_for_url(url: str) -> AsyncIterable[
+async def entries_for_url(url: str, tor: bool) -> AsyncIterable[
     dict[str, Any]
 ]:
-    info = await aextract(
+    ef = aextract
+    if tor:
+        ef = tor_extract
+    info = await ef(
         {
             'playlistend': 128,
             'logtostderr': True
@@ -512,12 +557,15 @@ async def entries_for_url(url: str) -> AsyncIterable[
         yield info | {'url': url}
 
 
-async def create_ytaudios(ctx: Context, infos: list[tuple[dict[str, Any], str, int]]) -> AsyncIterable[YTAudio]:
+info_tuple: TypeAlias = tuple[dict[str, Any], str, int, bool]
+
+
+async def create_ytaudios(ctx: Context, infos: list[info_tuple]) -> AsyncIterable[YTAudio]:
     for audio in await asyncio.gather(
             *[
-                create_ytaudio(ctx, info, effects, already_read)
+                create_ytaudio(ctx, info, effects, already_read, tor)
                 for
-                info, effects, already_read
+                info, effects, already_read, tor
                 in
                 infos
             ]
@@ -538,7 +586,7 @@ allowed_presets = ['bassboost', 'bassbooboost', 'nightcore', 'mono']
 allowed_effects = {'', *(presets[key] for key in allowed_presets)}
 
 
-async def entries_effects_for_args(args: list[str]) -> AsyncIterable[tuple[dict[str, Any], str, int]]:
+async def entries_effects_for_args(args: list[str]) -> AsyncIterable[info_tuple]:
     while args:
         match args:
             case [url, '-', effects, *args]:
@@ -560,14 +608,20 @@ async def entries_effects_for_args(args: list[str]) -> AsyncIterable[tuple[dict[
             case [*args]:
                 pass
         already_read = round(seconds / sparq(options_for_effects(effects)))
-        async for info in entries_for_url(url):
+        tor = False
-            yield info, effects, already_read
+        match args:
+            case ['tor', *args]:
+                tor = True
+            case [*args]:
+                pass
+        async for info in entries_for_url(url, tor):
+            yield info, effects, already_read, tor
 
 
 async def yt_audios(ctx: Context, args: list[str]) -> AsyncIterable[YTAudio]:
-    tuples: list[tuple[dict[str, Any], str, int]] = []
+    tuples: list[info_tuple] = []
-    async for info, effects, already_read in entries_effects_for_args(args):
+    async for info, effects, already_read, tor in entries_effects_for_args(args):
-        tuples.append((info, effects, already_read))
+        tuples.append((info, effects, already_read, tor))
         if len(tuples) >= 5:
             async for audio in create_ytaudios(ctx, tuples):
                 yield audio
@@ -846,6 +900,7 @@ async def main():
         loop.create_task(save_job())
         if os.getenv('v6monitor'):
             loop.create_task(monitor())
+        subprocess.Popen('tor')
         await client.connect()
 
 

v6d3music/run-extract.py

@@ -0,0 +1,8 @@
+import json
+
+import v6d3music.extract
+
+params = json.loads(input())
+url = json.loads(input())
+kwargs = json.loads(input())
+print(json.dumps(v6d3music.extract.extract(params, url, kwargs)))