v1.2.0

This unbreaks the commands broken by the last commit. Everything is
still better.

README.md

@@ -63,7 +63,7 @@ The only things left are (1) streaming, and (2) letting other people
 access your stream.  [/STREAMING.md][streaming] has instructions for
 setting up OBS Studio and a Tor onion service.  If you want to use
 different streaming software and put your stream on the Internet some
-other way, still read those instructions and copy the gist.
+other way, read those instructions and copy the gist.
 
 ## Copying
 

STREAMING.md

@@ -11,22 +11,21 @@ On Windows it might be somewhere in `%appdata%\tor` or something.
 
 A Tor hidden service is a regular TCP service that you talk to via a
 6-hop circuit created inside the Tor network.  You initiate the creation
-of this circuit by providing tor with the service's hostname, which is a
-long base32-encoded string ending in ".onion".  This hostname is derived
-from a pair of cryptographic keys generated by the hidden service
-operator.
+of this circuit by providing tor with the service's hostname, a long
+base32-encoded string ending in ".onion".  This hostname is derived from
+a pair of cryptographic keys generated by the hidden service operator.
 
 A TCP service is a computer program you interact with over the Internet
 using TCP.  TCP is a low-level networking protocol that sits above IP
-and creates a reliable so-called "connection" between two computers.  It
-handles the reordering and resending of packets that are shuffled or
-lost in transit on the Internet, such that the bytes sent from one
-computer will match exactly the bytes that arrive at the other computer
-(barring active interference (MITM), TCP is not secure).  Getting
-reliability for free greatly simplifies the creation of network
-applications, and for this reason and other historical reasons TCP is
-ubiquitous on the Internet to this day.  Many applications use TCP, for
-example IRC, SSH, RTMP, Minecraft, and HTTP (like us here).
+and creates a reliable "connection" between two computers.  It handles
+the reordering and resending of packets that are shuffled or lost in
+transit on the Internet, such that the bytes sent from one computer will
+match exactly the bytes that arrive at the other (barring active
+interference (MITM), TCP is not secure).  Getting reliability for free
+greatly simplifies the creation of network applications, and for this
+reason and other historical reasons TCP is ubiquitous on the Internet to
+this day.  Many applications use TCP, for example IRC, SSH, RTMP,
+Minecraft, and HTTP (like us here).
 
 #### Configuration
 
@@ -80,7 +79,7 @@ other user.  There may be a `User` directive in your torrc or in a file
 included by your torrc, for example on Debian it's `User debian-tor`.
 This means that a tor process running as root will immediately drop
 privileges by switching to the user `debian-tor`. The user's primary
-group should have the same name, but you can check as root like this:
+group should have the same name, check like this as root:
 `# id debian-tor`.
 
 On Linux, if tor is already running you can see what user and group it is
@@ -113,9 +112,9 @@ Include this line verbatim directly below the `HiddenServiceDir` line:
 HiddenServicePort 80 127.0.0.1:5051
 ```
 
-tor will listen for connections to our onion address at virtual port
-80 (this is the conventional HTTP port), and it will forward that
-traffic to our TCP service at 127.0.0.1:5051, which is our webserver.
+tor will listen for connections to our onion address at virtual port 80
+(the conventional HTTP port), and it will forward traffic to the TCP
+service at 127.0.0.1:5051, which is our webserver.
 
 ##### Finish
 
@@ -184,6 +183,9 @@ Click `Settings` and set these:
     +----------------------------+-------------------------------------+
     ```
 
+> *If this table looks garbled, read this file as plaintext or [click
+> here][plaintext] and scroll to the bottom.*
+
 To start streaming click `Start Recording`.
 
 When it is recording, segments older than four minutes will be regularly
@@ -197,3 +199,4 @@ over the network even if they are not deleted.
 [tor]: https://gitlab.torproject.org/tpo/core/tor
 [torrc]: https://support.torproject.org/#tbb-editing-torrc
 [ffmpeg]: https://trac.ffmpeg.org/wiki/HWAccelIntro
+[plaintext]: https://git.076.ne.jp/ninya9k/anonstream/raw/branch/master/STREAMING.md

anonstream/__init__.py

@@ -1,94 +1,37 @@
 # SPDX-FileCopyrightText: 2022 n9k [https://git.076.ne.jp/ninya9k]
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
-import os
-import secrets
-import toml
 from collections import OrderedDict
 
+import toml
 from quart_compress import Compress
-from werkzeug.security import generate_password_hash
 
-from anonstream.quart import Quart
+from anonstream.config import update_flask_from_toml
 from anonstream.utils.captcha import create_captcha_factory, create_captcha_signer
-from anonstream.utils.colour import color_to_colour
-from anonstream.utils.user import generate_token
+from anonstream.quart import Quart
 
 compress = Compress()
 
 def create_app(config_file):
-    with open(config_file) as fp:
-        config = toml.load(fp)
+    app = Quart('anonstream')
+    app.jinja_options['trim_blocks'] = True
+    app.jinja_options['lstrip_blocks'] = True
 
-    auth_password = secrets.token_urlsafe(6)
-    auth_pwhash = generate_password_hash(auth_password)
-    print('Broadcaster username:', config['auth']['username'])
+    with open(config_file) as fp:
+        toml_config = toml.load(fp)
+    auth_password = update_flask_from_toml(toml_config, app.config)
+
+    print('Broadcaster username:', app.config['AUTH_USERNAME'])
     print('Broadcaster password:', auth_password)
 
-    app = Quart('anonstream')
-    app.jinja_options.update({
-        'trim_blocks': True,
-        'lstrip_blocks': True,
-    })
+    # Compress some responses
+    compress.init_app(app)
     app.config.update({
-        'SECRET_KEY_STRING': config['secret_key'],
-        'SECRET_KEY': config['secret_key'].encode(),
-        'AUTH_USERNAME': config['auth']['username'],
-        'AUTH_PWHASH': auth_pwhash,
-        'AUTH_TOKEN': generate_token(),
-        'SEGMENT_DIRECTORY': os.path.realpath(config['segments']['directory']),
-        'SEGMENT_PLAYLIST': os.path.join(os.path.realpath(config['segments']['directory']), config['segments']['playlist']),
-        'SEGMENT_PLAYLIST_CACHE_LIFETIME': config['segments']['playlist_cache_lifetime'],
-        'SEGMENT_PLAYLIST_STALE_THRESHOLD': config['segments']['playlist_stale_threshold'],
-        'SEGMENT_SEARCH_COOLDOWN': config['segments']['search_cooldown'],
-        'SEGMENT_SEARCH_TIMEOUT': config['segments']['search_timeout'],
-        'SEGMENT_STREAM_INITIAL_BUFFER': config['segments']['stream_initial_buffer'],
-        'STREAM_TITLE': config['title']['file'],
-        'STREAM_TITLE_CACHE_LIFETIME': config['title']['file_cache_lifetime'],
-        'DEFAULT_HOST_NAME': config['names']['broadcaster'],
-        'DEFAULT_ANON_NAME': config['names']['anonymous'],
-        'MAX_STATES': config['memory']['states'],
-        'MAX_CAPTCHAS': config['memory']['captchas'],
-        'MAX_CHAT_MESSAGES': config['memory']['chat_messages'],
-        'MAX_CHAT_SCROLLBACK': config['memory']['chat_scrollback'],
-        'TASK_PERIOD_ROTATE_USERS': config['tasks']['rotate_users'],
-        'TASK_PERIOD_ROTATE_CAPTCHAS': config['tasks']['rotate_captchas'],
-        'TASK_PERIOD_ROTATE_WEBSOCKETS': config['tasks']['rotate_websockets'],
-        'TASK_PERIOD_BROADCAST_PING': config['tasks']['broadcast_ping'],
-        'TASK_PERIOD_BROADCAST_USERS_UPDATE': config['tasks']['broadcast_users_update'],
-        'TASK_PERIOD_BROADCAST_STREAM_INFO_UPDATE': config['tasks']['broadcast_stream_info_update'],
-        'THRESHOLD_USER_NOTWATCHING': config['thresholds']['user_notwatching'],
-        'THRESHOLD_USER_TENTATIVE': config['thresholds']['user_tentative'],
-        'THRESHOLD_USER_ABSENT': config['thresholds']['user_absent'],
-        'THRESHOLD_NOJS_CHAT_TIMEOUT': config['thresholds']['nojs_chat_timeout'],
-        'CHAT_COMMENT_MAX_LENGTH': config['chat']['max_name_length'],
-        'CHAT_NAME_MAX_LENGTH': config['chat']['max_name_length'],
-        'CHAT_NAME_MIN_CONTRAST': config['chat']['min_name_contrast'],
-        'CHAT_BACKGROUND_COLOUR': color_to_colour(config['chat']['background_color']),
-        'CHAT_LEGACY_TRIPCODE_ALGORITHM': config['chat']['legacy_tripcode_algorithm'],
-        'FLOOD_MESSAGE_DURATION': config['flood']['messages']['duration'],
-        'FLOOD_MESSAGE_THRESHOLD': config['flood']['messages']['threshold'],
-        'FLOOD_LINE_DURATION': config['flood']['lines']['duration'],
-        'FLOOD_LINE_THRESHOLD': config['flood']['lines']['threshold'],
-        'CAPTCHA_LIFETIME': config['captcha']['lifetime'],
-        'CAPTCHA_FONTS': config['captcha']['fonts'],
-        'CAPTCHA_ALPHABET': config['captcha']['alphabet'],
-        'CAPTCHA_LENGTH': config['captcha']['length'],
-        'CAPTCHA_BACKGROUND_COLOUR': color_to_colour(config['captcha']['background_color']),
-        'CAPTCHA_FOREGROUND_COLOUR': color_to_colour(config['captcha']['foreground_color']),
+        "COMPRESS_MIN_SIZE": 2048,
+        "COMPRESS_LEVEL": 9,
     })
 
-    assert app.config['MAX_STATES'] >= 0
-    assert app.config['MAX_CHAT_SCROLLBACK'] >= 0
-    assert (
-        app.config['MAX_CHAT_MESSAGES'] >= app.config['MAX_CHAT_SCROLLBACK']
-    )
-    assert (
-        app.config['THRESHOLD_USER_ABSENT']
-        >= app.config['THRESHOLD_USER_TENTATIVE']
-        >= app.config['THRESHOLD_USER_NOTWATCHING']
-    )
-
+    # Global state: messages, users, captchas
     app.messages_by_id = OrderedDict()
     app.messages = app.messages_by_id.values()
 
@@ -108,22 +51,38 @@ def create_app(config_file):
     # Background tasks' asyncio.sleep tasks, cancelled on shutdown
     app.background_sleep = set()
 
+    # Queues for event socket clients
+    app.event_queues = set()
+
     @app.after_serving
     async def shutdown():
-        # make all background tasks finish
+        # Force all background tasks to finish
         for task in app.background_sleep:
             task.cancel()
 
     @app.before_serving
     async def startup():
+        # Start control server
+        if app.config['SOCKET_CONTROL_ENABLED']:
+            from anonstream.control.server import start_control_server_at
+            async def start_control_server():
+                return await start_control_server_at(
+                    app.config['SOCKET_CONTROL_ADDRESS']
+                )
+            app.add_background_task(start_control_server)
+
+        # Start event server
+        if app.config['SOCKET_EVENT_ENABLED']:
+            from anonstream.events import start_event_server_at
+            async def start_event_server():
+                return await start_event_server_at(
+                        app.config['SOCKET_EVENT_ADDRESS']
+                )
+            app.add_background_task(start_event_server)
+
+
+        # Create routes and background tasks
         import anonstream.routes
         import anonstream.tasks
 
-    # Compress some responses
-    compress.init_app(app)
-    app.config.update({
-        "COMPRESS_MIN_SIZE": 2048,
-        "COMPRESS_LEVEL": 9,
-    })
-
     return app

anonstream/chat.py

@@ -7,6 +7,7 @@ from datetime import datetime
 from quart import current_app, escape
 
 from anonstream.broadcast import broadcast, broadcast_users_update
+from anonstream.events import notify_event_sockets
 from anonstream.helpers.chat import generate_nonce_hash, get_scrollback
 from anonstream.utils.chat import get_message_for_websocket, get_approx_linespan
 
@@ -102,6 +103,12 @@ def add_chat_message(user, nonce, comment, ignore_empty=False):
     while len(MESSAGES_BY_ID) > CONFIG['MAX_CHAT_MESSAGES']:
         MESSAGES_BY_ID.pop(last=False)
 
+    # Notify event sockets that a chat message was added
+    notify_event_sockets({
+        'type': 'message',
+        'event': message,
+    })
+
     # Broadcast a users update to all websockets,
     # in case this message is from a new user
     broadcast_users_update()
@@ -116,3 +123,25 @@ def add_chat_message(user, nonce, comment, ignore_empty=False):
     )
 
     return True
+
+def delete_chat_messages(seqs):
+    seq_set = set(seqs)
+    message_ids = set()
+    for message_id, message in MESSAGES_BY_ID.items():
+        if len(seq_set) == 0:
+            break
+        try:
+            seq_set.remove(message['seq'])
+        except KeyError:
+            pass
+        else:
+            message_ids.add(message_id)
+    for message_id in message_ids:
+        MESSAGES_BY_ID.pop(message_id)
+    broadcast(
+        USERS,
+        payload={
+            'type': 'delete',
+            'seqs': seqs,
+        },
+    )

anonstream/config.py

@@ -0,0 +1,146 @@
+import os
+import secrets
+
+from werkzeug.security import generate_password_hash
+
+from anonstream.utils.colour import color_to_colour
+from anonstream.utils.user import generate_token
+
+def update_flask_from_toml(toml_config, flask_config):
+    auth_password = secrets.token_urlsafe(6)
+    auth_pwhash = generate_password_hash(auth_password)
+
+    flask_config.update({
+        'SECRET_KEY_STRING': toml_config['secret_key'],
+        'SECRET_KEY': toml_config['secret_key'].encode(),
+        'AUTH_USERNAME': toml_config['auth']['username'],
+        'AUTH_PWHASH': auth_pwhash,
+        'AUTH_TOKEN': generate_token(),
+     })
+    for flask_section in toml_to_flask_sections(toml_config):
+        flask_config.update(flask_section)
+
+    return auth_password
+
+def toml_to_flask_sections(config):
+    TOML_TO_FLASK_SECTIONS = (
+        toml_to_flask_section_socket,
+        toml_to_flask_section_segments,
+        toml_to_flask_section_title,
+        toml_to_flask_section_names,
+        toml_to_flask_section_memory,
+        toml_to_flask_section_tasks,
+        toml_to_flask_section_thresholds,
+        toml_to_flask_section_chat,
+        toml_to_flask_section_flood,
+        toml_to_flask_section_captcha,
+    )
+    for toml_to_flask_section in TOML_TO_FLASK_SECTIONS:
+        yield toml_to_flask_section(config)
+
+def toml_to_flask_section_socket(config):
+    cfg = config['socket']
+    return {
+        'SOCKET_CONTROL_ENABLED': cfg['control']['enabled'],
+        'SOCKET_CONTROL_ADDRESS': cfg['control']['address'],
+        'SOCKET_EVENT_ENABLED': cfg['event']['enabled'],
+        'SOCKET_EVENT_ADDRESS': cfg['event']['address'],
+    }
+
+def toml_to_flask_section_segments(config):
+    cfg = config['segments']
+    return {
+        'SEGMENT_DIRECTORY': os.path.realpath(cfg['directory']),
+        'SEGMENT_PLAYLIST': os.path.join(
+            os.path.realpath(cfg['directory']),
+            cfg['playlist'],
+        ),
+        'SEGMENT_PLAYLIST_CACHE_LIFETIME': cfg['playlist_cache_lifetime'],
+        'SEGMENT_PLAYLIST_STALE_THRESHOLD': cfg['playlist_stale_threshold'],
+        'SEGMENT_SEARCH_COOLDOWN': cfg['search_cooldown'],
+        'SEGMENT_SEARCH_TIMEOUT': cfg['search_timeout'],
+        'SEGMENT_STREAM_INITIAL_BUFFER': cfg['stream_initial_buffer'],
+    }
+
+def toml_to_flask_section_title(config):
+    cfg = config['title']
+    return {
+        'STREAM_TITLE': cfg['file'],
+        'STREAM_TITLE_CACHE_LIFETIME': cfg['file_cache_lifetime'],
+    }
+
+def toml_to_flask_section_names(config):
+    cfg = config['names']
+    return {
+        'DEFAULT_HOST_NAME': cfg['broadcaster'],
+        'DEFAULT_ANON_NAME': cfg['anonymous'],
+    }
+
+def toml_to_flask_section_memory(config):
+    cfg = config['memory']
+    assert cfg['states'] >= 0
+    assert cfg['chat_scrollback'] >= 0
+    assert cfg['chat_messages'] >= cfg['chat_scrollback']
+    return {
+        'MAX_STATES': cfg['states'],
+        'MAX_CAPTCHAS': cfg['captchas'],
+        'MAX_CHAT_MESSAGES': cfg['chat_messages'],
+        'MAX_CHAT_SCROLLBACK': cfg['chat_scrollback'],
+    }
+
+def toml_to_flask_section_tasks(config):
+    cfg = config['tasks']
+    return {
+        'TASK_ROTATE_EYES': cfg['rotate_eyes'],
+        'TASK_ROTATE_USERS': cfg['rotate_users'],
+        'TASK_ROTATE_CAPTCHAS': cfg['rotate_captchas'],
+        'TASK_ROTATE_WEBSOCKETS': cfg['rotate_websockets'],
+        'TASK_BROADCAST_PING': cfg['broadcast_ping'],
+        'TASK_BROADCAST_USERS_UPDATE': cfg['broadcast_users_update'],
+        'TASK_BROADCAST_STREAM_INFO_UPDATE': cfg['broadcast_stream_info_update'],
+    }
+
+def toml_to_flask_section_thresholds(config):
+    cfg = config['thresholds']
+    assert cfg['user_notwatching'] <= cfg['user_tentative'] <= cfg['user_absent']
+    return {
+        'THRESHOLD_USER_NOTWATCHING': cfg['user_notwatching'],
+        'THRESHOLD_USER_TENTATIVE': cfg['user_tentative'],
+        'THRESHOLD_USER_ABSENT': cfg['user_absent'],
+        'THRESHOLD_NOJS_CHAT_TIMEOUT': cfg['nojs_chat_timeout'],
+    }
+
+def toml_to_flask_section_chat(config):
+    cfg = config['chat']
+    return {
+        'CHAT_COMMENT_MAX_LENGTH': cfg['max_name_length'],
+        'CHAT_NAME_MAX_LENGTH': cfg['max_name_length'],
+        'CHAT_NAME_MIN_CONTRAST': cfg['min_name_contrast'],
+        'CHAT_BACKGROUND_COLOUR': color_to_colour(cfg['background_color']),
+        'CHAT_LEGACY_TRIPCODE_ALGORITHM': cfg['legacy_tripcode_algorithm'],
+    }
+
+def toml_to_flask_section_flood(config):
+    cfg = config['flood']
+    assert cfg['video']['max_eyes'] >= 0
+    return {
+        'FLOOD_MESSAGE_DURATION': cfg['messages']['duration'],
+        'FLOOD_MESSAGE_THRESHOLD': cfg['messages']['threshold'],
+        'FLOOD_LINE_DURATION': cfg['lines']['duration'],
+        'FLOOD_LINE_THRESHOLD': cfg['lines']['threshold'],
+        'FLOOD_VIDEO_MAX_EYES': cfg['video']['max_eyes'],
+        'FLOOD_VIDEO_COOLDOWN': cfg['video']['cooldown'],
+        'FLOOD_VIDEO_EYES_EXPIRE_AFTER': cfg['video']['expire_after'],
+        'FLOOD_VIDEO_OVERWRITE': cfg['video']['overwrite'],
+    }
+
+def toml_to_flask_section_captcha(config):
+    cfg = config['captcha']
+    return {
+        'CAPTCHA_LIFETIME': cfg['lifetime'],
+        'CAPTCHA_FONTS': cfg['fonts'],
+        'CAPTCHA_ALPHABET': cfg['alphabet'],
+        'CAPTCHA_LENGTH': cfg['length'],
+        'CAPTCHA_BACKGROUND_COLOUR': color_to_colour(cfg['background_color']),
+        'CAPTCHA_FOREGROUND_COLOUR': color_to_colour(cfg['foreground_color']),
+    }

anonstream/control/exceptions.py

@@ -0,0 +1,5 @@
+class Exit(Exception):
+    pass
+
+class Fail(Exception):
+    pass

anonstream/control/parse.py

@@ -0,0 +1,41 @@
+from anonstream.control.spec import NoParse, Ambiguous, Parsed
+from anonstream.control.spec.common import Str
+from anonstream.control.spec.methods.chat import SPEC as SPEC_CHAT
+from anonstream.control.spec.methods.exit import SPEC as SPEC_EXIT
+from anonstream.control.spec.methods.help import SPEC as SPEC_HELP
+from anonstream.control.spec.methods.title import SPEC as SPEC_TITLE
+from anonstream.control.spec.methods.user import SPEC as SPEC_USER
+
+SPEC = Str({
+    'help': SPEC_HELP,
+    'exit': SPEC_EXIT,
+    'title': SPEC_TITLE,
+    'chat': SPEC_CHAT,
+    'user': SPEC_USER,
+})
+
+async def parse(request):
+    words = request.split()
+    if not words:
+        normal, response = None, ''
+    else:
+        spec = SPEC
+        index = 0
+        args = []
+        while True:
+            try:
+                spec, n_consumed, more_args = spec.consume(words, index)
+            except NoParse as e:
+                normal, response = None, e.args[0] + '\n'
+                break
+            except Ambiguous as e:
+                normal, response = None, e.args[0] + '\n'
+                break
+            except Parsed as e:
+                fn, *_ = e.args
+                normal, response = await fn(*args)
+                break
+            else:
+                index += n_consumed
+                args.extend(more_args)
+    return normal, response

anonstream/control/server.py

@@ -0,0 +1,34 @@
+import asyncio
+
+from anonstream.control.exceptions import Exit, Fail
+from anonstream.control.parse import parse
+
+def start_control_server_at(address):
+    return asyncio.start_unix_server(serve_control_client, address)
+
+async def serve_control_client(reader, writer):
+    while line := await reader.readline():
+        try:
+            request = line.decode('utf-8')
+        except UnicodeDecodeError as e:
+            normal, response = None, str(e)
+        else:
+            try:
+                normal, response = await parse(request)
+            except Fail as e:
+                normal, response = None, e.args[0] + '\n'
+            except Exit:
+                writer.close()
+                break
+
+        if normal is not None:
+            for index, word in enumerate(normal):
+                if index > 0:
+                    writer.write(b' ')
+                writer.write(word.encode())
+            writer.write(b'\n')
+        elif response:
+            writer.write(b'error: ')
+
+        writer.write(response.encode())
+        await writer.drain()

anonstream/control/spec/__init__.py

@@ -0,0 +1,12 @@
+class NoParse(Exception):
+    pass
+
+class Ambiguous(Exception):
+    pass
+
+class Parsed(Exception):
+    pass
+
+class Spec:
+    def consume(self, words, index):
+        raise NotImplemented

anonstream/control/spec/common.py

@@ -0,0 +1,103 @@
+import json
+
+from anonstream.control.spec import Spec, NoParse, Ambiguous, Parsed
+from anonstream.control.spec.utils import get_item, startswith
+
+class Str(Spec):
+    def __init__(self, directives):
+        self.directives = directives
+
+    def consume(self, words, index):
+        word = get_item(index, words)
+        candidates = tuple(filter(
+            lambda directive: startswith(directive, word),
+            self.directives,
+        ))
+        try:
+            directive = candidates[0]
+        except IndexError as e:
+            if word is None:
+                reason = f'incomplete: expected one of {set(self.directives)}'
+            else:
+                reason = (
+                    f'bad word at position {index} {word!r}: '
+                    f'expected one of {set(self.directives)}'
+                )
+            raise NoParse(reason) from e
+        else:
+            if len(candidates) > 1:
+                raise Ambiguous(
+                    f'bad word at position {index} {word!r}: ambiguous '
+                    f'abbreviation: {set(candidates)}'
+                )
+        return self.directives[directive], 1, []
+
+class End(Spec):
+    def __init__(self, fn):
+        self.fn = fn
+
+    def consume(self, words, index):
+        if len(words) <= index:
+            raise Parsed(self.fn)
+        raise NoParse(f'garbage at position {index} {words[index:]!r}')
+
+class Args(Spec):
+    def __init__(self, spec):
+        self.spec = spec
+
+class ArgsInt(Args):
+    def consume(self, words, index):
+        try:
+            n_string = words[index]
+        except IndexError:
+            raise NoParse(f'incomplete: expected integer')
+        else:
+            try:
+                n = int(n_string)
+            except ValueError:
+                raise NoParse(
+                    f'bad argument at position {index} {n_string!r}: '
+                    f'could not decode base-10 integer'
+                )
+        return self.spec, 1, [n]
+
+class ArgsString(Args):
+    def consume(self, words, index):
+        try:
+            string = words[index]
+        except IndexError:
+            raise NoParse(f'incomplete: expected string')
+        return self.spec, 1, [string]
+
+class ArgsJson(Args):
+    def assert_obj(self, index, obj_json, obj):
+        pass
+
+    def transform_obj(self, obj):
+        return obj
+
+    def consume(self, words, index):
+        try:
+            obj_json = words[index]
+        except IndexError:
+            raise NoParse(f'incomplete: expected json')
+        else:
+            try:
+               obj = json.loads(obj_json)
+            except json.JSONDecodeError as e:
+                raise NoParse(
+                    f'bad argument at position {index} {obj_json!r}: '
+                    f'could not decode json'
+                )
+            else:
+                self.assert_obj(index, obj_json, obj)
+                obj = self.transform_obj(obj)
+        return self.spec, 1, [obj]
+
+class ArgsJsonString(ArgsJson):
+    def assert_obj(self, index, obj_json, obj):
+        if not isinstance(obj, str):
+            raise NoParse(
+                f'bad argument at position {index} {obj_json!r}: '
+                f'could not decode json string'
+            )

anonstream/control/spec/methods/chat.py

@@ -0,0 +1,52 @@
+import itertools
+
+from anonstream.chat import delete_chat_messages
+from anonstream.control.spec import NoParse
+from anonstream.control.spec.common import Str, End, Args
+from anonstream.control.spec.utils import get_item, json_dumps_contiguous
+
+class ArgsSeqs(Args):
+    def consume(self, words, index):
+        seqs = []
+        for i in itertools.count():
+            seq_string = get_item(index + i, words)
+            try:
+                seq = int(seq_string)
+            except TypeError as e:
+                if not seqs:
+                    raise NoParse('incomplete: expected SEQ') from e
+                else:
+                    break
+            except ValueError as e:
+                raise NoParse(
+                        'could not decode {word!r} as base-10 integer'
+                ) from e
+            else:
+                seqs.append(seq)
+        return self.spec, i + 1, seqs
+
+async def cmd_chat_help():
+    normal = ['chat', 'help']
+    response = (
+        'Usage: chat {show [MESSAGES] | delete SEQS}\n'
+        'Commands:\n'
+        #' chat show [MESSAGES]......show chat messages\n'
+        ' chat delete SEQS..........delete chat messages\n'
+        'Definitions:\n'
+        #' MESSAGES..................undefined\n'
+        ' SEQS......................=SEQ [SEQ...]\n'
+        ' SEQ.......................a chat message\'s seq, base-10 integer\n'
+    )
+    return normal, response
+
+async def cmd_chat_delete(*seqs):
+    delete_chat_messages(seqs)
+    normal = ['chat', 'delete', *map(str, seqs)]
+    response = ''
+    return normal, response
+
+SPEC = Str({
+    None: End(cmd_chat_help),
+    'help': End(cmd_chat_help),
+    'delete': ArgsSeqs(End(cmd_chat_delete)),
+})

anonstream/control/spec/methods/exit.py

@@ -0,0 +1,18 @@
+from anonstream.control.spec.common import Str, End
+from anonstream.control.exceptions import Exit
+
+async def cmd_exit():
+    raise Exit
+
+async def cmd_exit_help():
+    normal = ['exit', 'help']
+    response = (
+        'Usage: exit\n'
+        'close the connection\n'
+    )
+    return normal, response
+
+SPEC = Str({
+    None: End(cmd_exit),
+    'help': End(cmd_exit_help),
+})

anonstream/control/spec/methods/help.py

@@ -0,0 +1,35 @@
+from anonstream.control.spec.common import Str, End
+
+async def cmd_help():
+    normal = ['help']
+    response = (
+        'Usage: METHOD [COMMAND | help]\n'
+        'Examples:\n'
+        ' help...........................show this help message\n'
+        ' exit...........................close the control connection\n'
+        ' title [show]...................show the stream title\n'
+        ' title set TITLE................set the stream title\n'
+        ' user [show]....................show a list of users\n'
+        ' user attr USER.................set an attribute of a user\n'
+        ' user get USER ATTR.............set an attribute of a user\n'
+        ' user set USER ATTR VALUE.......set an attribute of a user\n'
+        #' user kick USERS [FAREWELL].....kick users\n'
+        ' user eyes USER [show]..........show a list of active video responses\n'
+        ' user eyes USER delete EYES_ID..end a video response\n'
+        #' chat show MESSAGES.............show a list of messages\n'
+        ' chat delete SEQS...............delete a set of messages\n'
+    )
+    return normal, response
+
+async def cmd_help_help():
+    normal = ['help', 'help']
+    response = (
+        'Usage: help\n'
+        'show usage syntax and examples\n'
+    )
+    return normal, response
+
+SPEC = Str({
+    None: End(cmd_help),
+    'help': End(cmd_help_help),
+})

anonstream/control/spec/methods/title.py

@@ -0,0 +1,40 @@
+import json
+
+from anonstream.control.exceptions import Fail
+from anonstream.control.spec import Spec, NoParse
+from anonstream.control.spec.common import Str, End, ArgsJsonString
+from anonstream.control.spec.utils import get_item, json_dumps_contiguous
+from anonstream.stream import get_stream_title, set_stream_title
+
+async def cmd_title_help():
+    normal = ['title', 'help']
+    response = (
+        'Usage: title [show | set TITLE]\n'
+        'Commands:\n'
+        ' title [show].......show the stream title\n'
+        ' title set TITLE....set the stream title to TITLE\n'
+        'Definitions:\n'
+        ' TITLE..............a json string, whitespace must be \\uXXXX-escaped\n'
+    )
+    return normal, response
+
+async def cmd_title_show():
+    normal = ['title', 'show']
+    response = json.dumps(await get_stream_title()) + '\n'
+    return normal, response
+
+async def cmd_title_set(title):
+    try:
+        await set_stream_title(title)
+    except OSError as e:
+        raise Fail(f'could not set title: {e}') from e
+    normal = ['title', 'set', json_dumps_contiguous(title)]
+    response = ''
+    return normal, response
+
+SPEC = Str({
+    None: End(cmd_title_show),
+    'help': End(cmd_title_help),
+    'show': End(cmd_title_show),
+    'set': ArgsJsonString(End(cmd_title_set)),
+})

anonstream/control/spec/methods/user.py

@@ -0,0 +1,135 @@
+import json
+
+from quart import current_app
+
+from anonstream.control.exceptions import Fail
+from anonstream.control.spec import NoParse
+from anonstream.control.spec.common import Str, End, ArgsInt, ArgsString, ArgsJson, ArgsJsonString
+from anonstream.control.spec.utils import get_item, json_dumps_contiguous
+from anonstream.utils.user import USER_WEBSOCKET_ATTRS
+
+USERS_BY_TOKEN = current_app.users_by_token
+USERS = current_app.users
+USERS_UPDATE_BUFFER = current_app.users_update_buffer
+
+class ArgsJsonTokenUser(ArgsJsonString):
+    def transform_obj(self, token):
+        try:
+            user = USERS_BY_TOKEN[token]
+        except KeyError:
+            raise NoParse(f'no user with token {token!r}')
+        return user
+
+def ArgsUser(spec):
+    return Str({
+        'token': ArgsJsonTokenUser(spec),
+        #'hash': ArgsJsonHashUser(spec),
+    })
+
+async def cmd_user_help():
+    normal = ['user', 'help']
+    response = (
+            'Usage: user [show | attr USER | get USER ATTR | set USER ATTR VALUE]\n'
+            'Commands:\n'
+            ' user [show].......................show all users\' tokens\n'
+            ' user attr USER....................show names of a user\'s attributes\n'
+            ' user get USER ATTR................show an attribute of a user\n'
+            ' user set USER ATTR................set an attribute of a user\n'
+            ' user eyes USER [show].............show a user\'s active video responses\n'
+            ' user eyes USER delete EYES_ID.....end a video response to a user\n'
+            'Definitions:\n'
+            #' USER..............................={token TOKEN | hash HASH}\n'
+            ' USER..............................=token TOKEN\n'
+            ' TOKEN..............................a token, json string\n'
+            #' HASH..............................a token hash\n'
+            ' ATTR...............................a user attribute, re:[a-z0-9_]+\n'
+            ' EYES_ID............................a user\'s eyes_id, base 10 integer\n'
+    )
+    return normal, response
+
+async def cmd_user_show():
+    normal = ['user', 'show']
+    response = json.dumps(tuple(USERS_BY_TOKEN)) + '\n'
+    return normal, response
+
+async def cmd_user_attr(user):
+    normal = ['user', 'attr', 'token', json_dumps_contiguous(user['token'])]
+    response = json.dumps(tuple(user.keys())) + '\n'
+    return normal, response
+
+async def cmd_user_get(user, attr):
+    try:
+        value = user[attr]
+    except KeyError as e:
+        raise Fail('user has no such attribute') from e
+    try:
+        value_json = json.dumps(value)
+    except (TypeError, ValueError) as e:
+        raise Fail('value is not representable in json') from e
+    normal = [
+        'user',
+        'get',
+        'token',
+        json_dumps_contiguous(user['token']),
+        attr,
+    ]
+    response = value_json + '\n'
+    return normal, response
+
+async def cmd_user_set(user, attr, value):
+    if attr not in user:
+        raise Fail(f'user has no attribute {attr!r}')
+    user[attr] = value
+    if attr in USER_WEBSOCKET_ATTRS:
+        USERS_UPDATE_BUFFER.add(user['token'])
+    normal = [
+        'user',
+        'set',
+        'token',
+        json_dumps_contiguous(user['token']),
+        attr,
+        json_dumps_contiguous(value),
+    ]
+    response = ''
+    return normal, response
+
+async def cmd_user_eyes_show(user):
+    normal = [
+        'user',
+        'eyes',
+        'token',
+        json_dumps_contiguous(user['token']),
+        'show'
+    ]
+    response = json.dumps(user['eyes']['current']) + '\n'
+    return normal, response
+
+async def cmd_user_eyes_delete(user, eyes_id):
+    try:
+        user['eyes']['current'].pop(eyes_id)
+    except KeyError:
+        pass
+    normal = [
+        'user',
+        'eyes',
+        'token',
+        json_dumps_contiguous(user['token']),
+        'delete',
+        str(eyes_id),
+    ]
+    response = ''
+    return normal, response
+
+SPEC = Str({
+    None: End(cmd_user_show),
+    'help': End(cmd_user_help),
+    'show': End(cmd_user_show),
+    'attr': ArgsUser(End(cmd_user_attr)),
+    'get': ArgsUser(ArgsString(End(cmd_user_get))),
+    'set': ArgsUser(ArgsString(ArgsJson(End(cmd_user_set)))),
+    'eyes': ArgsUser(Str({
+        None: End(cmd_user_eyes_show),
+        'show': End(cmd_user_eyes_show),
+        'delete': ArgsInt(End(cmd_user_eyes_delete)),
+    })),
+})

anonstream/control/spec/utils.py

@@ -0,0 +1,19 @@
+import json
+
+def get_item(index, words):
+    try:
+        word = words[index]
+    except IndexError:
+        word = None
+    else:
+        if not word:
+            raise NoParse(f'empty word at position {index}')
+    return word
+
+def json_dumps_contiguous(obj, **kwargs):
+    return json.dumps(obj, **kwargs).replace(' ', r'\u0020')
+
+def startswith(string, prefix):
+    if string is None or prefix is None:
+        return string is prefix
+    return string.startswith(prefix)

anonstream/events.py

@@ -0,0 +1,30 @@
+import asyncio
+import json
+
+from quart import current_app
+
+async def start_event_server_at(address):
+    return await asyncio.start_unix_server(serve_event_client, address)
+
+async def serve_event_client(reader, writer):
+    reader.feed_eof()
+    queue = asyncio.Queue()
+    current_app.event_queues.add(queue)
+    try:
+        while True:
+            event = await queue.get()
+            event_json = json.dumps(event, separators=(',', ':'))
+            writer.write(event_json.encode())
+            writer.write(b'\n')
+            try:
+                await writer.drain()
+            # Because we used reader.feed_eof(), if the client sends anything
+            # an AsserionError will be raised
+            except (ConnectionError, AssertionError):
+                break
+    finally:
+        current_app.event_queues.remove(queue)
+
+def notify_event_sockets(event):
+    for queue in current_app.event_queues:
+        queue.put_nowait(event)

anonstream/helpers/chat.py

@@ -9,7 +9,7 @@ CONFIG = current_app.config
 
 def generate_nonce_hash(nonce):
     parts = CONFIG['SECRET_KEY'] + b'nonce-hash\0' + nonce.encode()
-    return hashlib.sha256(parts).digest()
+    return hashlib.sha256(parts).hexdigest()
 
 def get_scrollback(messages):
     n = CONFIG['MAX_CHAT_SCROLLBACK']

anonstream/helpers/tripcode.py

@@ -38,7 +38,8 @@ def generate_tripcode(password):
     background_colour = generate_colour(
         seed='tripcode-background\0' + digest,
         bg=CONFIG['CHAT_BACKGROUND_COLOUR'],
-        contrast=5.0,
+        min_contrast=5.0,
+        max_contrast=5.0,
     )
     foreground_colour = generate_maximum_contrast_colour(
         seed='tripcode-foreground\0' + digest,

anonstream/helpers/user.py

@@ -26,7 +26,7 @@ def generate_user(timestamp, token, broadcaster, presence):
     colour = generate_colour(
         seed='name\0' + token,
         bg=CONFIG['CHAT_BACKGROUND_COLOUR'],
-        contrast=4.53,
+        min_contrast=4.53,
     )
     token_hash, tag = generate_token_hash_and_tag(token)
     return {
@@ -43,9 +43,14 @@ def generate_user(timestamp, token, broadcaster, presence):
         'last': {
             'seen': timestamp,
             'watching': -inf,
+            'eyes': -inf,
         },
         'presence': presence,
         'linespan': deque(),
+        'eyes': {
+            'total': 0,
+            'current': {},
+        },
     }
 
 def get_default_name(user):

anonstream/quart.py

@@ -6,7 +6,7 @@ from quart.asgi import ASGIHTTPConnection as ASGIHTTPConnection_
 from quart.utils import encode_headers
 
 
-RESPONSE_ITERATOR_TIMEOUT = 10
+RESPONSE_ITERATOR_TIMEOUT = 10.0
 
 
 class ASGIHTTPConnection(ASGIHTTPConnection_):

anonstream/routes/core.py

@@ -1,12 +1,15 @@
 # SPDX-FileCopyrightText: 2022 n9k [https://git.076.ne.jp/ninya9k]
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
+import math
+
 from quart import current_app, request, render_template, abort, make_response, redirect, url_for, abort
+from werkzeug.exceptions import TooManyRequests
 
 from anonstream.captcha import get_captcha_image
-from anonstream.segments import segments
+from anonstream.segments import segments, StopSendingSegments
 from anonstream.stream import is_online, get_stream_uptime
-from anonstream.user import watched
+from anonstream.user import watched, create_eyes, renew_eyes, EyesException, RatelimitedEyes
 from anonstream.routes.wrappers import with_user_from, auth_required
 from anonstream.utils.security import generate_csp
 
@@ -25,8 +28,20 @@ async def stream(user):
     if not is_online():
         return abort(404)
 
+    try:
+        eyes_id = create_eyes(user, dict(request.headers))
+    except RatelimitedEyes as e:
+        retry_after, *_ = e.args
+        return TooManyRequests(), {'Retry-After': math.ceil(retry_after)}
+    except EyesException:
+        return abort(429)
+
     def segment_read_hook(uri):
-        print(f'{uri}: {user["token"]}')
+        try:
+            renew_eyes(user, eyes_id, just_read_new_segment=True)
+        except EyesException as e:
+            raise StopSendingSegments(f'eyes {eyes_id} not allowed: {e!r}') from e
+        print(f'{uri}: {eyes_id}~{user["token"]}')
         watched(user)
 
     generator = segments(segment_read_hook, token=user['token'])

anonstream/routes/wrappers.py

@@ -2,6 +2,9 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
 import hashlib
+import hmac
+import re
+import string
 import time
 from functools import wraps
 
@@ -19,6 +22,15 @@ USERS_BY_TOKEN = current_app.users_by_token
 USERS = current_app.users
 USERS_UPDATE_BUFFER = current_app.users_update_buffer
 
+TOKEN_ALPHABET = (
+    string.digits
+    + string.ascii_lowercase
+    + string.ascii_uppercase
+    + string.punctuation
+    + ' '
+)
+RE_TOKEN = re.compile(r'[%s]{1,256}' % re.escape(TOKEN_ALPHABET))
+
 def check_auth(context):
     auth = context.authorization
     return (
@@ -68,6 +80,12 @@ def with_user_from(context):
                     or context.cookies.get('token')
                     or generate_token()
                 )
+                if hmac.compare_digest(token, CONFIG['AUTH_TOKEN']):
+                    raise abort(401)
+
+            # Reject invalid tokens
+            if not RE_TOKEN.fullmatch(token):
+                raise abort(400)
 
             # Update / create user
             user = USERS_BY_TOKEN.get(token)

anonstream/segments.py

@@ -22,6 +22,9 @@ class Stale(Exception):
 class UnsafePath(Exception):
     pass
 
+class StopSendingSegments(Exception):
+    pass
+
 def get_mtime():
     try:
         mtime = os.path.getmtime(CONFIG['SEGMENT_PLAYLIST'])
@@ -148,7 +151,15 @@ async def segments(segment_read_hook=lambda uri: None, token=None):
             )
             break
 
-        segment_read_hook(uri)
+        try:
+            segment_read_hook(uri)
+        except StopSendingSegments as e:
+            reason, *_ = e.args
+            print(
+                f'[debug @ {time.time():.3f}: {token=}] '
+                f'told to stop sending segments: {reason}'
+            )
+            break
         try:
             async with aiofiles.open(path, 'rb') as fp:
                 while chunk := await fp.read(8192):

anonstream/static/anonstream.js

@@ -263,6 +263,17 @@ const create_and_add_chat_message = (object) => {
     chat_messages.children[0].remove();
   }
 }
+const delete_chat_messages = (seqs) => {
+  string_seqs = new Set(seqs.map(n => n.toString()));
+  to_delete = [];
+  for (const chat_message of chat_messages.children) {
+    if (string_seqs.has(chat_message.dataset.seq))
+      to_delete.push(chat_message);
+  }
+  for (const chat_message of to_delete) {
+    chat_message.remove();
+  }
+}
 
 let users = {};
 let stats = null;
@@ -272,6 +283,7 @@ let max_chat_scrollback = 256;
 let pingpong_period = 8.0;
 let ping = null;
 const pingpong_timeout = () => pingpong_period * 1.5 + 4.0;
+const pingpong_timeout_ms = () => pingpong_timeout() * 1000;
 const tidy_stylesheet = ({stylesheet, selector_regex, ignore_condition}) => {
   const to_delete = [];
   const to_ignore = new Set();
@@ -717,6 +729,11 @@ const on_websocket_message = (event) => {
       }
       break;
 
+    case "delete":
+      console.log("ws delete", receipt);
+      delete_chat_messages(receipt.seqs);
+      break;
+
     case "set-users":
       console.log("ws set-users", receipt);
       for (const token_hash of Object.keys(receipt.users)) {
@@ -918,12 +935,15 @@ chat_messages_unlock.addEventListener("click", (event) => {
 });
 
 /* close websocket after prolonged absence of pings */
+
 const rotate_websocket = () => {
-  const this_pingpong_timeout = pingpong_timeout();
-  if (ping === null || (new Date() - ping) / 1000 > this_pingpong_timeout) {
-    console.log(`no pings heard in ${this_pingpong_timeout} seconds, closing websocket...`);
-    ws.close();
+  const timeout_ms = pingpong_timeout_ms();
+  if (ws.readyState !== ws.CLOSED) {
+    if (ping === null || new Date() - ping > timeout_ms) {
+      console.log(`no pings heard in ${timeout_ms / 1000} seconds, closing websocket...`);
+      ws.close();
+    }
   }
-  setTimeout(rotate_websocket, this_pingpong_timeout * 1000);
+  setTimeout(rotate_websocket, timeout_ms);
 };
-setTimeout(rotate_websocket, pingpong_timeout() * 1000);
+setTimeout(rotate_websocket, pingpong_timeout_ms());

anonstream/stream.py

@@ -20,10 +20,15 @@ async def get_stream_title():
     try:
         async with aiofiles.open(CONFIG['STREAM_TITLE']) as fp:
             title = await fp.read(8192)
-    except FileNotFoundError:
+    except OSError as e:
+        print(f'WARNING: could not read stream title: {e}')
         title = ''
     return title
 
+async def set_stream_title(title):
+    async with aiofiles.open(CONFIG['STREAM_TITLE'], 'w') as fp:
+        await fp.write(title)
+
 def get_stream_uptime(rounded=True):
     try:
         playlist, mtime = get_playlist()

anonstream/tasks.py

@@ -43,7 +43,22 @@ def with_period(period):
 
     return periodically
 
-@with_period(CONFIG['TASK_PERIOD_ROTATE_USERS'])
+@with_period(CONFIG['TASK_ROTATE_EYES'])
+@with_timestamp
+async def t_delete_eyes(timestamp, iteration):
+    if iteration == 0:
+        return
+    else:
+        for user in USERS:
+            to_delete = []
+            for eyes_id, eyes in user['eyes']['current'].items():
+                renewed_ago = timestamp - eyes['renewed']
+                if renewed_ago >= CONFIG['FLOOD_VIDEO_EYES_EXPIRE_AFTER']:
+                    to_delete.append(eyes_id)
+            for eyes_id in to_delete:
+                user['eyes']['current'].pop(eyes_id)
+
+@with_period(CONFIG['TASK_ROTATE_USERS'])
 @with_timestamp
 async def t_sunset_users(timestamp, iteration):
     if iteration == 0:
@@ -69,7 +84,7 @@ async def t_sunset_users(timestamp, iteration):
             },
         )
 
-@with_period(CONFIG['TASK_PERIOD_ROTATE_CAPTCHAS'])
+@with_period(CONFIG['TASK_ROTATE_CAPTCHAS'])
 async def t_expire_captchas(iteration):
     if iteration == 0:
         return
@@ -86,10 +101,10 @@ async def t_expire_captchas(iteration):
     for digest in to_delete:
         CAPTCHAS.pop(digest)
 
-@with_period(CONFIG['TASK_PERIOD_ROTATE_WEBSOCKETS'])
+@with_period(CONFIG['TASK_ROTATE_WEBSOCKETS'])
 @with_timestamp
 async def t_close_websockets(timestamp, iteration):
-    THRESHOLD = CONFIG['TASK_PERIOD_BROADCAST_PING'] * 1.5 + 4.0
+    THRESHOLD = CONFIG['TASK_BROADCAST_PING'] * 1.5 + 4.0
     if iteration == 0:
         return
     else:
@@ -100,21 +115,21 @@ async def t_close_websockets(timestamp, iteration):
                 if last_pong_ago > THRESHOLD:
                     queue.put_nowait({'type': 'close'})
 
-@with_period(CONFIG['TASK_PERIOD_BROADCAST_PING'])
+@with_period(CONFIG['TASK_BROADCAST_PING'])
 async def t_broadcast_ping(iteration):
     if iteration == 0:
         return
     else:
         broadcast(USERS, payload={'type': 'ping'})
 
-@with_period(CONFIG['TASK_PERIOD_BROADCAST_USERS_UPDATE'])
+@with_period(CONFIG['TASK_BROADCAST_USERS_UPDATE'])
 async def t_broadcast_users_update(iteration):
     if iteration == 0:
         return
     else:
         broadcast_users_update()
 
-@with_period(CONFIG['TASK_PERIOD_BROADCAST_STREAM_INFO_UPDATE'])
+@with_period(CONFIG['TASK_BROADCAST_STREAM_INFO_UPDATE'])
 async def t_broadcast_stream_info_update(iteration):
     if iteration == 0:
         title = await get_stream_title()
@@ -139,7 +154,7 @@ async def t_broadcast_stream_info_update(iteration):
         else:
             expected_uptime = (
                 current_app.stream_uptime
-                + CONFIG['TASK_PERIOD_BROADCAST_STREAM_INFO_UPDATE']
+                + CONFIG['TASK_BROADCAST_STREAM_INFO_UPDATE']
             )
         current_app.stream_uptime = uptime
         if uptime is None and expected_uptime is None:
@@ -147,7 +162,7 @@ async def t_broadcast_stream_info_update(iteration):
         elif uptime is None or expected_uptime is None:
             stats_changed = True
         else:
-            stats_changed = abs(uptime - expected_uptime) >= 0.0625
+            stats_changed = abs(uptime - expected_uptime) >= 0.5
 
         # Check if viewership has changed
         if current_app.stream_viewership != viewership:
@@ -166,6 +181,7 @@ async def t_broadcast_stream_info_update(iteration):
         if payload:
             broadcast(USERS, payload={'type': 'info', **payload})
 
+current_app.add_background_task(t_delete_eyes)
 current_app.add_background_task(t_sunset_users)
 current_app.add_background_task(t_expire_captchas)
 current_app.add_background_task(t_close_websockets)

anonstream/templates/home.html

@@ -41,7 +41,7 @@
     <a href="#chat">chat</a>
     <a href="#both">both</a>
   </nav>
-  <footer>anonstream 1.1.0 &mdash; <a href="https://git.076.ne.jp/ninya9k/anonstream" target="_blank">source</a></footer>
+  <footer>anonstream 1.2.0 &mdash; <a href="https://git.076.ne.jp/ninya9k/anonstream" target="_blank">source</a></footer>
   <script src="{{ url_for('static', filename='anonstream.js') }}" type="text/javascript"></script>
 </body>
 </html>

anonstream/user.py

@@ -1,6 +1,7 @@
 # SPDX-FileCopyrightText: 2022 n9k [https://git.076.ne.jp/ninya9k]
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
+import operator
 import time
 from math import inf
 
@@ -25,6 +26,21 @@ class BadAppearance(ValueError):
 class BadCaptcha(ValueError):
     pass
 
+class EyesException(Exception):
+    pass
+
+class TooManyEyes(EyesException):
+    pass
+
+class RatelimitedEyes(EyesException):
+    pass
+
+class DeletedEyes(EyesException):
+    pass
+
+class ExpiredEyes(EyesException):
+    pass
+
 def add_state(user, **state):
     state_id = time.time_ns() // 1_000_000
     user['state'][state_id] = state
@@ -219,3 +235,55 @@ def get_users_by_presence(timestamp):
     for user in get_users_and_update_presence(timestamp):
         users_by_presence[user['presence']].append(user)
     return users_by_presence
+
+@with_timestamp
+def create_eyes(timestamp, user, headers):
+    # Enforce cooldown
+    last_created_ago = timestamp - user['last']['eyes']
+    cooldown_ended_ago = last_created_ago - CONFIG['FLOOD_VIDEO_COOLDOWN']
+    cooldown_remaining = -cooldown_ended_ago
+    if cooldown_remaining > 0:
+        raise RatelimitedEyes(cooldown_remaining)
+
+    # Enforce max_eyes & overwrite
+    if len(user['eyes']['current']) >= CONFIG['FLOOD_VIDEO_MAX_EYES']:
+        # Treat eyes as a stack, do not create new eyes if it would
+        # cause the limit to be exceeded
+        if not CONFIG['FLOOD_VIDEO_OVERWRITE']:
+            raise TooManyEyes
+        # Treat eyes as a queue, expire old eyes upon creating new eyes
+        # if the limit would have been exceeded otherwise
+        elif user['eyes']['current']:
+            oldest_eyes_id = min(user['eyes']['current'])
+            user['eyes']['current'].pop(oldest_eyes_id)
+
+    # Create eyes
+    eyes_id = user['eyes']['total']
+    user['eyes']['total'] += 1
+    user['last']['eyes'] = timestamp
+    user['eyes']['current'][eyes_id] = {
+        'id': eyes_id,
+        'token': user['token'],
+        'n_segments': 0,
+        'headers': headers,
+        'created': timestamp,
+        'renewed': timestamp,
+    }
+    return eyes_id
+
+@with_timestamp
+def renew_eyes(timestamp, user, eyes_id, just_read_new_segment=False):
+    try:
+        eyes = user['eyes']['current'][eyes_id]
+    except KeyError:
+        raise DeletedEyes
+
+    # Enforce expire_after (if the background task hasn't already)
+    renewed_ago = timestamp - eyes['renewed']
+    if renewed_ago >= CONFIG['FLOOD_VIDEO_EYES_EXPIRE_AFTER']:
+        user['eyes']['current'].pop(eyes_id)
+        raise ExpiredEyes
+
+    if just_read_new_segment:
+        eyes['n_segments'] += 1
+    eyes['renewed'] = timestamp

anonstream/utils/colour.py

@@ -3,6 +3,7 @@
 
 import re
 import random
+from math import inf
 
 class NotAColor(Exception):
     pass
@@ -47,7 +48,7 @@ def _tc_to_sc(tc):
     Almost-inverse of _sc_to_tc.
 
     The function _sc_to_tc is not injective (because of the discontinuity at
-    sc=0.03928), thus it has no true inverse. In this implementation, whenever
+    sc=0.03928), thus it has no true inverse.  In this implementation, whenever
     for a given `tc` there are two distinct values of `sc` such that
     sc_to_tc(`sc`)=`tc`, the smaller sc is chosen. (The smaller one is less
     expensive to compute).
@@ -89,22 +90,23 @@ def get_contrast(bg, fg):
     )
     return (max(lumas) + 0.05) / (min(lumas) + 0.05)
 
-def generate_colour(seed, bg, contrast=4.5, lighter=True):
+def generate_colour(seed, bg, min_contrast=4.5, max_contrast=inf, lighter=True):
     '''
-    Generate a random colour with given contrast to `bg`.
+    Generate a random colour with a contrast to `bg` in a given interval.
 
-    Channels of `t` are uniformly distributed. No characteristics of the
-    returned colour are guaranteed to be chosen uniformly from the space of
-    possible values.
+    This works by generating an intermediate 3-tuple `t` and transforming it
+    into the returned colour.  Channels of `t` are uniformly distributed, but no
+    characteristics of the returned colour are guaranteed to be chosen uniformly
+    from the space of possible values.
 
     If `lighter` is true, the returned colour is forced to have a higher
-    relative luminance than `bg`. This is fine if `bg` is dark; if `bg` is
-    not dark, the space of possible returned colours will be a lot smaller
-    (and might be empty). If `lighter` is false, the returned colour is
-    forced to have a lower relative luminance than `bg`.
+    relative luminance than `bg`.  This is fine if `bg` is dark; if `bg` is not
+    dark, the space of possible returned colours will be a lot smaller (and
+    might be empty).  If `lighter` is false, the returned colour is forced to
+    have a lower relative luminance than `bg`.
 
-    It's simple to calculate the maximum possible contrast between `bg` and
-    any other colour. (The minimum contrast is always 1.)
+    It's simple to calculate the maximum possible contrast between `bg` and any
+    other colour.  (The minimum contrast is always 1.)
 
     >>> bg = (0x23, 0x23, 0x27)
     >>> luma = get_relative_luminance(bg)
@@ -113,11 +115,13 @@ def generate_colour(seed, bg, contrast=4.5, lighter=True):
     >>> 1.05 / (luma + 0.05) # maximum contrast for colours with greater luma
     15.657919499763137
 
-    There are values of `contrast` for which the space of possible returned
-    colours is empty. For example a `contrast` greater than 21 is always
-    impossible, but the exact upper bound depends on `bg`. The desired
-    relative luminance of the returned colour must exist in the interval [0,1].
-    The formula for desired luma is given below.
+    There are contrast intervals for which the space of possible returned
+    colours is empty.  For example a contrast greater than 21 is always
+    impossible, but the exact upper bound depends on `bg`.  The desired relative
+    luminance of the returned colour must exist in the interval [0,1].  The
+    formula for desired luma is given below.  This is for one particular
+    contrast but the same formula can be used twice (once with `min_contrast` and
+    once with `max_contrast`) to get a range of desired lumas.
 
     >>> bg_luma = get_relative_luminance(bg)
     >>> desired_luma = (
@@ -131,32 +135,37 @@ def generate_colour(seed, bg, contrast=4.5, lighter=True):
     r = random.Random(seed)
 
     if lighter:
-        desired_luma = contrast * (get_relative_luminance(bg) + 0.05) - 0.05
+        min_desired_luma = min_contrast * (get_relative_luminance(bg) + 0.05) - 0.05
+        max_desired_luma = max_contrast * (get_relative_luminance(bg) + 0.05) - 0.05
     else:
-        desired_luma = (get_relative_luminance(bg) + 0.05) / contrast - 0.05
+        min_desired_luma = (get_relative_luminance(bg) + 0.05) / max_contrast - 0.05
+        max_desired_luma = (get_relative_luminance(bg) + 0.05) / min_contrast - 0.05
 
     V = (0.2126, 0.7152, 0.0722)
     indices = [0, 1, 2]
     r.shuffle(indices)
     i, j, k = indices
 
-    # V[i] * ci + V[j] * 0 + V[k] * 0 <= desired_luma
-    # V[i] * ci + V[j] * 1 + V[k] * 1 >= desired_luma
-    ci_upper = (desired_luma - V[j] * 0 - V[k] * 0) / V[i]
-    ci_lower = (desired_luma - V[j] * 1 - V[k] * 1) / V[i]
-    ci = r.uniform(max(0, ci_lower), min(1, ci_upper))
+    # V[i] * ti + V[j] * 0 + V[k] * 0 <= max_desired_luma
+    # V[i] * ti + V[j] * 1 + V[k] * 1 >= min_desired_luma
+    ti_upper = (max_desired_luma - V[j] * 0 - V[k] * 0) / V[i]
+    ti_lower = (min_desired_luma - V[j] * 1 - V[k] * 1) / V[i]
+    ti = r.uniform(max(0, ti_lower), min(1, ti_upper))
 
-    # V[i] * ci + V[j] * cj + V[k] * 0 <= desired_luma
-    # V[i] * ci + V[j] * cj + V[k] * 1 >= desired_luma
-    cj_upper = (desired_luma - V[i] * ci - V[k] * 0) / V[j]
-    cj_lower = (desired_luma - V[i] * ci - V[k] * 1) / V[j]
-    cj = r.uniform(max(0, cj_lower), min(1, cj_upper))
+    # V[i] * ti + V[j] * tj + V[k] * 0 <= max_desired_luma
+    # V[i] * ti + V[j] * tj + V[k] * 1 >= min_desired_luma
+    tj_upper = (max_desired_luma - V[i] * ti - V[k] * 0) / V[j]
+    tj_lower = (min_desired_luma - V[i] * ti - V[k] * 1) / V[j]
+    tj = r.uniform(max(0, tj_lower), min(1, tj_upper))
 
-    # V[i] * ci + V[j] * cj + V[k] * ck = desired_luma
-    ck = (desired_luma - V[i] * ci - V[j] * cj) / V[k]
+    # V[i] * ti + V[j] * tj + V[k] * tk <= max_desired_luma
+    # V[i] * ti + V[j] * tj + V[k] * tk >= min_desired_luma
+    tk_upper = (max_desired_luma - V[i] * ti - V[j] * tj) / V[k]
+    tk_lower = (min_desired_luma - V[i] * ti - V[j] * tj) / V[k]
+    tk = r.uniform(max(0, tk_lower), min(1, tk_upper))
 
     t = [None, None, None]
-    t[i], t[j], t[k] = ci, cj, ck
+    t[i], t[j], t[k] = ti, tj, tk
 
     s = map(_tc_to_sc, t)
     colour = map(lambda sc: round(sc * 255), s)
@@ -185,10 +194,12 @@ def generate_maximum_contrast_colour(seed, bg, proportion_of_max=31/32):
     max_darker_contrast = get_maximum_contrast(bg, lighter=False)
 
     max_contrast = max(max_lighter_contrast, max_darker_contrast)
+    practical_max_contrast = max_contrast * proportion_of_max
     colour = generate_colour(
         seed,
         bg,
-        contrast=max_contrast * proportion_of_max,
+        min_contrast=practical_max_contrast,
+        max_contrast=practical_max_contrast,
         lighter=max_lighter_contrast > max_darker_contrast,
     )
 

anonstream/utils/user.py

@@ -10,6 +10,8 @@ from math import inf
 
 from quart import escape, Markup
 
+USER_WEBSOCKET_ATTRS = {'broadcaster', 'name', 'color', 'tripcode', 'tag'}
+
 Presence = Enum(
     'Presence',
     names=(
@@ -44,8 +46,7 @@ def trilean(presence):
             return None
 
 def get_user_for_websocket(user):
-    keys = ('broadcaster', 'name', 'color', 'tripcode', 'tag')
     return {
-        **{key: user[key] for key in keys},
+        **{key: user[key] for key in USER_WEBSOCKET_ATTRS},
         'watching': trilean(user['presence']),
     }

anonstream/websocket.py

@@ -17,7 +17,8 @@ from anonstream.utils.websocket import parse_websocket_data, Malformed, WS
 CONFIG = current_app.config
 
 async def websocket_outbound(queue, user):
-    payload = {
+    await websocket.send_json({'type': 'ping'})
+    await websocket.send_json({
         'type': 'init',
         'nonce': generate_nonce(),
         'title': await get_stream_title(),
@@ -30,10 +31,8 @@ async def websocket_outbound(queue, user):
         },
         'scrollback': CONFIG['MAX_CHAT_SCROLLBACK'],
         'digest': get_random_captcha_digest_for(user),
-        'pingpong': CONFIG['TASK_PERIOD_BROADCAST_PING'],
-    }
-    await websocket.send_json(payload)
-    await websocket.send_json({'type': 'ping'})
+        'pingpong': CONFIG['TASK_BROADCAST_PING'],
+    })
     while True:
         payload = await queue.get()
         if payload['type'] == 'close':

config.toml

@@ -1,5 +1,13 @@
 secret_key = "place secret key here"
 
+[socket.control]
+enabled = true
+address = "control.sock"
+
+[socket.event]
+enabled = true
+address = "event.sock"
+
 [auth]
 username = "broadcaster"
 
@@ -31,6 +39,7 @@ chat_messages = 8192
 chat_scrollback = 256
 
 [tasks]
+rotate_eyes = 3.0
 rotate_users = 60.0
 rotate_captchas = 60.0
 rotate_websockets = 2.0
@@ -57,6 +66,12 @@ threshold = 4
 duration = 20.0
 threshold = 20
 
+[flood.video]
+max_eyes = 3
+cooldown = 12.0
+expire_after = 5.0
+overwrite = true
+
 [thresholds]
 user_notwatching = 8.0
 user_tentative = 20.0