Merge branch 'dev-allowedness'

anonstream/__init__.py

@@ -6,8 +6,9 @@ from collections import OrderedDict
 from quart_compress import Compress
 
 from anonstream.config import update_flask_from_toml
-from anonstream.utils.captcha import create_captcha_factory, create_captcha_signer
 from anonstream.quart import Quart
+from anonstream.utils.captcha import create_captcha_factory, create_captcha_signer
+from anonstream.utils.user import generate_blank_allowedness
 
 __version__ = '1.3.6'
 
@@ -30,7 +31,7 @@ def create_app(toml_config):
         'COMPRESS_LEVEL': 9,
     })
 
-    # Global state: messages, users, captchas
+    # Global state: messages, users, captchas, etc.
     app.messages_by_id = OrderedDict()
     app.messages = app.messages_by_id.values()
 
@@ -41,7 +42,8 @@ def create_app(toml_config):
     app.captcha_factory = create_captcha_factory(app.config['CAPTCHA_FONTS'])
     app.captcha_signer = create_captcha_signer(app.config['SECRET_KEY'])
 
-    app.failures = OrderedDict()
+    app.failures = OrderedDict() # access captcha failures
+    app.allowedness = generate_blank_allowedness()
 
     # State for tasks
     app.users_update_buffer = set()

anonstream/control/parse.py

@@ -3,6 +3,7 @@
 
 from anonstream.control.spec import ParseException, Parsed
 from anonstream.control.spec.common import Str
+from anonstream.control.spec.methods.allowedness import SPEC as SPEC_ALLOWEDNESS
 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
@@ -15,6 +16,7 @@ SPEC = Str({
     'title': SPEC_TITLE,
     'chat': SPEC_CHAT,
     'user': SPEC_USER,
+    'allowednesss': SPEC_ALLOWEDNESS,
 })
 
 async def parse(request):

anonstream/control/spec/common.py

@@ -7,6 +7,8 @@ from anonstream.control.spec import Spec, NoParse, Ambiguous, Parsed
 from anonstream.control.spec.utils import get_item, startswith
 
 class Str(Spec):
+    AS_ARG = False
+
     def __init__(self, directives):
         self.directives = directives
 
@@ -33,7 +35,10 @@ class Str(Spec):
                     f'bad word at position {index} {word!r}: ambiguous '
                     f'abbreviation: {set(candidates)}'
                 )
-        return self.directives[directive], 1, []
+        args = []
+        if self.AS_ARG:
+            args.append(directive)
+        return self.directives[directive], 1, args
 
 class End(Spec):
     def __init__(self, fn):
@@ -48,6 +53,9 @@ class Args(Spec):
     def __init__(self, spec):
         self.spec = spec
 
+class ArgsStr(Str):
+    AS_ARG = True
+
 class ArgsInt(Args):
     def consume(self, words, index):
         try:
@@ -102,6 +110,14 @@ class ArgsJson(Args):
                 obj = self.transform_obj(obj)
         return self.spec, 1, [obj]
 
+class ArgsJsonBoolean(ArgsJson):
+    def assert_obj(self, index, obj_json, obj):
+        if not isinstance(obj, bool):
+            raise NoParse(
+                    f'bad argument at position {index} {obj_json!r}: '
+                    f'could not decode json boolean'
+                )
+
 class ArgsJsonString(ArgsJson):
     def assert_obj(self, index, obj_json, obj):
         if not isinstance(obj, str):
@@ -109,3 +125,24 @@ class ArgsJsonString(ArgsJson):
                 f'bad argument at position {index} {obj_json!r}: '
                 f'could not decode json string'
             )
+
+class ArgsJsonArray(ArgsJson):
+    def assert_obj(self, index, obj_json, obj):
+        if not isinstance(obj, list):
+            raise NoParse(
+                f'bad argument at position {index} {obj_json!r}: '
+                f'could not decode json array'
+            )
+
+class ArgsJsonStringArray(ArgsJson):
+    def assert_obj(self, index, obj_json, obj):
+        if not isinstance(obj, list):
+            raise NoParse(
+                f'bad argument at position {index} {obj_json!r}: '
+                f'could not decode json array'
+            )
+        if any(not isinstance(item, str) for item in obj):
+            raise NoParse(
+                f'bad argument at position {index} {obj_json!r}: '
+                f'could not decode json array of strings'
+            )

anonstream/control/spec/methods/allowedness.py

@@ -0,0 +1,105 @@
+# SPDX-FileCopyrightText: 2022 n9k <https://git.076.ne.jp/ninya9k>
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+import json
+
+from quart import current_app
+
+from anonstream.control.exceptions import CommandFailed
+from anonstream.control.spec.common import Str, End, ArgsStr, ArgsJsonBoolean, ArgsJsonString, ArgsJsonStringArray
+from anonstream.control.spec.utils import json_dumps_contiguous
+
+ALLOWEDNESS = current_app.allowedness
+
+async def cmd_allowedness_help():
+    normal = ['allowedness', 'help']
+    response = (
+        'Usage: allowedness [show | set default BOOLEAN | add LIST KEYTUPLE VALUE | remove LIST KEYTUPLE VALUE]\n'
+        'Commands:\n'
+        ' allowedness [show]........................\n'
+        ' allowedness setdefault BOOLEAN............set the default allowedness\n'
+        #' allowedness clear LIST all................\n'
+        #' allowedness clear LIST keytuple KEYTUPLE..\n'
+        ' allowedness add LIST KEYTUPLE STRING......add to the blacklist/whitelist\n'
+        ' allowedness remove LIST KEYTUPLE STRING...remove from the blacklist/whitelist\n'
+        'Definitions:\n'
+        ' BOOLEAN...................................={true | false}\n'
+        ' LIST......................................={blacklist | whitelist}\n'
+        ' KEYTUPLE..................................keys to burrow into a user with, e.g. (\'tripcode\', \'digest\'), encoded as a json array\n'
+        ' STRING....................................a json string\n'
+    )
+    return normal, response
+
+async def cmd_allowedness_show():
+    allowedness_for_json = {
+        'blacklist': {},
+        'whitelist': {},
+        'default': ALLOWEDNESS['default'],
+    }
+    for colourlist in ('blacklist', 'whitelist'):
+        for keytuple, values in ALLOWEDNESS[colourlist].items():
+            allowedness_for_json[colourlist]['.'.join(keytuple)] = sorted(values)
+    normal = ['allowedness', 'show']
+    response = json.dumps(allowedness_for_json) + '\n'
+    return normal, response
+
+async def cmd_allowedness_setdefault(value):
+    ALLOWEDNESS['default'] = value
+    normal = ['allowednesss', 'setdefault', json_dumps_contiguous(value)]
+    response = ''
+    return normal, response
+
+async def cmd_allowedness_add(colourlist, keytuple_list, value):
+    keytuple = tuple(keytuple_list)
+    try:
+        values = ALLOWEDNESS[colourlist][keytuple]
+    except KeyError:
+        raise CommandFailed(f'no such keytuple {keytuple!r} in list {colourlist!r}')
+    else:
+        values.add(value)
+    normal = [
+        'allowednesss',
+        'add',
+        colourlist,
+        json_dumps_contiguous(keytuple),
+        json_dumps_contiguous(value),
+    ]
+    response = ''
+    return normal, response
+
+async def cmd_allowedness_remove(colourlist, keytuple_list, value):
+    keytuple = tuple(keytuple_list)
+    try:
+        values = ALLOWEDNESS[colourlist][keytuple]
+    except KeyError:
+        raise CommandFailed(f'no such keytuple {keytuple!r} in list {colourlist!r}')
+    else:
+        try:
+            values.remove(value)
+        except KeyError:
+            pass
+    normal = [
+        'allowednesss',
+        'remove',
+        colourlist,
+        json_dumps_contiguous(keytuple),
+        json_dumps_contiguous(value),
+    ]
+    response = ''
+    return normal, response
+
+SPEC = Str({
+    None: End(cmd_allowedness_show),
+    'help': End(cmd_allowedness_help),
+    'show': End(cmd_allowedness_show),
+    'setdefault': ArgsJsonBoolean(End(cmd_allowedness_setdefault)),
+    #'clear': cmd_allowedness_clear,
+    'add': ArgsStr({
+        'blacklist': ArgsJsonStringArray(ArgsJsonString(End(cmd_allowedness_add))),
+        'whitelist': ArgsJsonStringArray(ArgsJsonString(End(cmd_allowedness_add))),
+    }),
+    'remove': ArgsStr({
+        'blacklist': ArgsJsonStringArray(ArgsJsonString(End(cmd_allowedness_remove))),
+        'whitelist': ArgsJsonStringArray(ArgsJsonString(End(cmd_allowedness_remove))),
+    }),
+})

anonstream/control/spec/methods/help.py

@@ -16,11 +16,14 @@ async def cmd_help():
         ' 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'
+        ' allowedness [show].............show the current allowedness\n'
+        ' allowedness setdefault BOOLEAN.set the default allowedness\n'
+        ' allowedness add SET STRING.....add to the blacklist/whitelist\n'
+        ' allowedness remove SET STRING..remove from the blacklist/whitelist\n'
     )
     return normal, response
 

anonstream/helpers/user.py

@@ -48,6 +48,7 @@ def generate_user(
             'watching': -inf,
             'eyes': -inf,
             'reading': -inf,
+            'allowed': -inf,
         },
         'presence': presence,
         'linespan': deque(),

anonstream/routes/core.py

@@ -4,41 +4,50 @@
 import math
 
 from quart import current_app, request, render_template, abort, make_response, redirect, url_for, send_from_directory
-from werkzeug.exceptions import NotFound, TooManyRequests
+from werkzeug.exceptions import Forbidden, NotFound, TooManyRequests
 
 from anonstream.access import add_failure, pop_failure
 from anonstream.captcha import get_captcha_image, get_random_captcha_digest
 from anonstream.segments import segments, StopSendingSegments
 from anonstream.stream import is_online, get_stream_uptime
-from anonstream.user import watching, create_eyes, renew_eyes, EyesException, RatelimitedEyes, TooManyEyes
+from anonstream.user import watching, create_eyes, renew_eyes, EyesException, RatelimitedEyes, TooManyEyes, ensure_allowedness, Blacklisted, SecretClub
 from anonstream.routes.wrappers import with_user_from, auth_required, clean_cache_headers, generate_and_add_user
 from anonstream.helpers.captcha import check_captcha_digest, Answer
 from anonstream.utils.security import generate_csp
 from anonstream.utils.user import identifying_string
+from anonstream.wrappers import with_timestamp
 
 CAPTCHA_SIGNER = current_app.captcha_signer
 STATIC_DIRECTORY = current_app.root_path / 'static'
 
 @current_app.route('/')
-@with_user_from(request, fallback_to_token=True)
+@with_user_from(request, fallback_to_token=True, ignore_allowedness=True)
 async def home(timestamp, user_or_token):
     match user_or_token:
-        case str() | None:
+        case str() | None as token:
             failure_id = request.args.get('failure', type=int)
             response = await render_template(
                 'captcha.html',
                 csp=generate_csp(),
-                token=user_or_token,
+                token=token,
                 digest=get_random_captcha_digest(),
                 failure=pop_failure(failure_id),
             )
-        case dict():
-            response = await render_template(
-                'home.html',
-                csp=generate_csp(),
-                user=user_or_token,
-                version=current_app.version,
-            )
+        case dict() as user:
+            try:
+                ensure_allowedness(user, timestamp=timestamp)
+            except Blacklisted:
+                raise Forbidden('You have been blacklisted.')
+            except SecretClub:
+                # TODO allow changing tripcode
+                raise Forbidden('You have not been whitelisted.')
+            else:
+                response = await render_template(
+                    'home.html',
+                    csp=generate_csp(),
+                    user=user,
+                    version=current_app.version,
+                )
     return response
 
 @current_app.route('/stream.mp4')
@@ -66,15 +75,18 @@ async def stream(timestamp, user):
                 f'End one of those before making a new request.'
             )
         else:
-            def segment_read_hook(uri):
+            @with_timestamp(precise=True)
+            def segment_read_hook(timestamp, uri):
+                user['last']['seen'] = timestamp
                 try:
-                    renew_eyes(user, eyes_id, just_read_new_segment=True)
+                    renew_eyes(timestamp, user, eyes_id, just_read_new_segment=True)
                 except EyesException as e:
                     raise StopSendingSegments(
                         f'eyes {eyes_id} not allowed: {e!r}'
                     ) from e
+                else:
+                    user['last']['watching'] = timestamp
                 print(f'{uri}: \033[37m{eyes_id}\033[0m~{identifying_string(user)}')
-                watching(user)
             generator = segments(segment_read_hook, token=f'\033[35m{user["token"]}\033[0m')
             response = await make_response(generator)
             response.headers['Content-Type'] = 'video/mp4'
@@ -97,11 +109,10 @@ async def captcha(timestamp, user_or_token):
         return image, {'Content-Type': 'image/jpeg'}
 
 @current_app.post('/access')
-@with_user_from(request, fallback_to_token=True)
+@with_user_from(request, fallback_to_token=True, ignore_allowedness=True)
 async def access(timestamp, user_or_token):
     match user_or_token:
-        case str() | None:
-            token = user_or_token
+        case str() | None as token:
             form = await request.form
             digest = form.get('digest', '')
             answer = form.get('answer', '')
@@ -118,8 +129,8 @@ async def access(timestamp, user_or_token):
             if failure_id is not None:
                 url = url_for('home', token=token, failure=failure_id)
                 raise abort(redirect(url, 303))
-        case dict():
-            user = user_or_token
+        case dict() as user:
+            pass
     url = url_for('home', token=user['token'])
     return redirect(url, 303)
 

anonstream/routes/websocket.py

@@ -4,26 +4,32 @@
 import asyncio
 from quart import current_app, websocket
 
-from anonstream.user import see
+from anonstream.user import see, ensure_allowedness, AllowednessException
 from anonstream.websocket import websocket_outbound, websocket_inbound
 from anonstream.routes.wrappers import with_user_from
 
 @current_app.websocket('/live')
-@with_user_from(websocket, fallback_to_token=True)
+@with_user_from(websocket, fallback_to_token=True, ignore_allowedness=True)
 async def live(timestamp, user_or_token):
     match user_or_token:
         case str() | None:
             await websocket.send_json({'type': 'kick'})
             await websocket.close(1001)
         case dict() as user:
-            queue = asyncio.Queue()
-            user['websockets'][queue] = timestamp
-            user['last']['reading'] = timestamp
-
-            producer = websocket_outbound(queue, user)
-            consumer = websocket_inbound(queue, user)
             try:
-                await asyncio.gather(producer, consumer)
-            finally:
-                see(user)
-                user['websockets'].pop(queue)
+                ensure_allowedness(user, timestamp=timestamp)
+            except AllowednessException:
+                await websocket.send_json({'type': 'kick'})
+                await websocket.close(1001)
+            else:
+                queue = asyncio.Queue()
+                user['websockets'][queue] = timestamp
+                user['last']['reading'] = timestamp
+
+                producer = websocket_outbound(queue, user)
+                consumer = websocket_inbound(queue, user)
+                try:
+                    await asyncio.gather(producer, consumer)
+                finally:
+                    see(user)
+                    user['websockets'].pop(queue)

anonstream/routes/wrappers.py

@@ -13,6 +13,7 @@ from werkzeug.exceptions import BadRequest, Unauthorized, Forbidden
 from werkzeug.security import check_password_hash
 
 from anonstream.broadcast import broadcast
+from anonstream.user import ensure_allowedness, Blacklisted, SecretClub
 from anonstream.helpers.user import generate_user
 from anonstream.utils.user import generate_token, Presence
 from anonstream.wrappers import get_timestamp
@@ -86,7 +87,7 @@ def generate_and_add_user(
     USERS_UPDATE_BUFFER.add(token)
     return user
 
-def with_user_from(context, fallback_to_token=False):
+def with_user_from(context, fallback_to_token=False, ignore_allowedness=False):
     def with_user_from_context(f):
         @wraps(f)
         async def wrapper(*args, **kwargs):
@@ -134,6 +135,8 @@ def with_user_from(context, fallback_to_token=False):
                 if user is not None:
                     user['last']['seen'] = timestamp
                     user['headers'] = tuple(context.headers)
+                    if not ignore_allowedness:
+                        assert_allowedness(timestamp, user)
                     response = await f(timestamp, user, *args, **kwargs)
                 elif fallback_to_token:
                     #assert not broadcaster
@@ -156,6 +159,8 @@ def with_user_from(context, fallback_to_token=False):
                         broadcaster,
                         headers=tuple(context.headers),
                     )
+                if not ignore_allowedness:
+                    assert_allowedness(timestamp, user)
                 response = await f(timestamp, user, *args, **kwargs)
 
             # Set cookie
@@ -207,3 +212,11 @@ def clean_cache_headers(f):
         return response
 
     return wrapper
+
+def assert_allowedness(timestamp, user):
+    try:
+        ensure_allowedness(user, timestamp=timestamp)
+    except Blacklisted as e:
+        raise Forbidden('You have been blacklisted.')
+    except SecretClub as e:
+        raise Forbidden('You have not been whitelisted.')

anonstream/tasks.py

@@ -9,7 +9,7 @@ from quart import current_app, websocket
 
 from anonstream.broadcast import broadcast, broadcast_users_update
 from anonstream.stream import is_online, get_stream_title, get_stream_uptime_and_viewership
-from anonstream.user import get_absent_users, get_sunsettable_users, deverify
+from anonstream.user import get_absent_users, get_sunsettable_users, deverify, ensure_allowedness, AllowednessException
 from anonstream.wrappers import with_timestamp
 
 CONFIG = current_app.config
@@ -116,6 +116,12 @@ async def t_close_websockets(timestamp, iteration):
     else:
         for user in USERS:
             for queue in user['websockets']:
+                # Check allowedness
+                try:
+                    ensure_allowedness(user, timestamp=timestamp)
+                except AllowednessException:
+                    queue.put_nowait({'type': 'kick'})
+                # Check expiry
                 last_pong = user['websockets'][queue]
                 last_pong_ago = timestamp - last_pong
                 if last_pong_ago > THRESHOLD:

anonstream/user.py

@@ -3,6 +3,7 @@
 
 import operator
 import time
+from functools import reduce
 from math import inf
 
 from quart import current_app
@@ -17,6 +18,7 @@ from anonstream.utils.user import get_user_for_websocket, trilean
 CONFIG = current_app.config
 MESSAGES = current_app.messages
 USERS = current_app.users
+ALLOWEDNESS = current_app.allowedness
 CAPTCHA_SIGNER = current_app.captcha_signer
 USERS_UPDATE_BUFFER = current_app.users_update_buffer
 
@@ -41,6 +43,18 @@ class DeletedEyes(EyesException):
 class ExpiredEyes(EyesException):
     pass
 
+class DisallowedEyes(EyesException):
+    pass
+
+class AllowednessException(Exception):
+    pass
+
+class Blacklisted(AllowednessException):
+    pass
+
+class SecretClub(AllowednessException):
+    pass
+
 def add_state(user, **state):
     state_id = time.time_ns() // 1_000_000
     user['state'][state_id] = state
@@ -253,6 +267,9 @@ def get_users_by_presence(timestamp):
 
 @with_timestamp(precise=True)
 def create_eyes(timestamp, user, headers):
+    # Unlike in renew_eyes, allowedness is NOT checked here because it is
+    # assumed to have already been checked (by the route handler).
+
     # Enforce cooldown
     last_created_ago = timestamp - user['last']['eyes']
     cooldown_ended_ago = last_created_ago - CONFIG['FLOOD_VIDEO_COOLDOWN']
@@ -286,7 +303,6 @@ def create_eyes(timestamp, user, headers):
     }
     return eyes_id
 
-@with_timestamp(precise=True)
 def renew_eyes(timestamp, user, eyes_id, just_read_new_segment=False):
     try:
         eyes = user['eyes']['current'][eyes_id]
@@ -299,6 +315,41 @@ def renew_eyes(timestamp, user, eyes_id, just_read_new_segment=False):
         user['eyes']['current'].pop(eyes_id)
         raise ExpiredEyes
 
+    # Ensure allowedness
+    try:
+        ensure_allowedness(user, timestamp=timestamp)
+    except AllowednessException as e:
+        user['eyes']['current'].pop(eyes_id)
+        raise DisallowedEyes from e
+
     if just_read_new_segment:
         eyes['n_segments'] += 1
     eyes['renewed'] = timestamp
+
+def ensure_allowedness(user, timestamp=None):
+    if timestamp is None:
+        timestamp = get_timestamp()
+
+    # Check against blacklist
+    for keytuple, values in ALLOWEDNESS['blacklist'].items():
+        try:
+            value = reduce(lambda mapping, key: mapping[key], keytuple, user)
+        except (KeyError, TypeError):
+            value = None
+        if value in values:
+            raise Blacklisted
+
+    # Check against whitelist
+    for keytuple, values in ALLOWEDNESS['whitelist'].items():
+        try:
+            value = reduce(lambda mapping, key: mapping[key], keytuple, user)
+        except (KeyError, TypeError):
+            value = None
+        if value in values:
+            break
+    else:
+        # Apply default
+        if not ALLOWEDNESS['default']:
+            raise SecretClub
+
+    user['last']['allowed'] = timestamp

anonstream/utils/user.py

@@ -60,3 +60,17 @@ def identifying_string(user, ansi=True):
         token_hash = f'\033[32m{token_hash}\033[0m'
         token = f'\033[35m{token}\033[0m'
     return '/'.join((tag, token_hash, token))
+
+def generate_blank_allowedness():
+    return {
+        'blacklist': {
+            ('token',): set(),
+            ('token_hash',): set(),
+        },
+        'whitelist': {
+            ('token',): set(),
+            ('token_hash',): set(),
+            ('tripcode', 'digest'): set(),
+        },
+        'default': True,
+    }

anonstream/websocket.py

@@ -9,7 +9,7 @@ from quart import current_app, websocket
 from anonstream.stream import get_stream_title, get_stream_uptime_and_viewership
 from anonstream.captcha import get_random_captcha_digest_for
 from anonstream.chat import get_all_messages_for_websocket, add_chat_message, Rejected
-from anonstream.user import get_all_users_for_websocket, see, reading, verify, deverify, BadCaptcha, try_change_appearance
+from anonstream.user import get_all_users_for_websocket, see, reading, verify, deverify, BadCaptcha, try_change_appearance, ensure_allowedness, AllowednessException
 from anonstream.wrappers import with_timestamp, get_timestamp
 from anonstream.utils.chat import generate_nonce
 from anonstream.utils.user import identifying_string
@@ -18,6 +18,9 @@ from anonstream.utils.websocket import parse_websocket_data, Malformed, WS
 CONFIG = current_app.config
 
 async def websocket_outbound(queue, user):
+    # This function does NOT check alllowedness at first, only later.
+    # Allowedness is assumed to be checked beforehand (by the route handler).
+    # These first two websocket messages are always sent.
     await websocket.send_json({'type': 'ping'})
     await websocket.send_json({
         'type': 'init',
@@ -36,14 +39,26 @@ async def websocket_outbound(queue, user):
     })
     while True:
         payload = await queue.get()
-        if payload['type'] == 'close':
+        if payload['type'] == 'kick':
+            await websocket.send_json(payload)
+            await websocket.close(1001)
+            break
+        elif payload['type'] == 'close':
             await websocket.close(1011)
             break
         else:
-            await websocket.send_json(payload)
+            try:
+                ensure_allowedness(user)
+            except AllowednessException:
+                websocket.send_json({'type': 'kick'})
+                await websocket.close(1001)
+                break
+            else:
+                await websocket.send_json(payload)
 
 async def websocket_inbound(queue, user):
     while True:
+        # Read from websocket
         try:
             receipt = await websocket.receive_json()
         except json.JSONDecodeError:
@@ -51,26 +66,34 @@ async def websocket_inbound(queue, user):
         finally:
             timestamp = get_timestamp()
             see(user, timestamp=timestamp)
-        try:
-            receipt_type, parsed = parse_websocket_data(receipt)
-        except Malformed as e:
-            error , *_ = e.args
-            payload = {
-                'type': 'error',
-                'because': error,
-            }
-        else:
-            match receipt_type:
-                case WS.MESSAGE:
-                    handle = handle_inbound_message
-                case WS.APPEARANCE:
-                    handle = handle_inbound_appearance
-                case WS.CAPTCHA:
-                    handle = handle_inbound_captcha
-                case WS.PONG:
-                    handle = handle_inbound_pong
-            payload = handle(timestamp, queue, user, *parsed)
 
+        # Prepare response
+        try:
+            ensure_allowedness(user)
+        except AllowednessException:
+            payload = {'type': 'kick'}
+        else:
+            try:
+                receipt_type, parsed = parse_websocket_data(receipt)
+            except Malformed as e:
+                error , *_ = e.args
+                payload = {
+                    'type': 'error',
+                    'because': error,
+                }
+            else:
+                match receipt_type:
+                    case WS.MESSAGE:
+                        handle = handle_inbound_message
+                    case WS.APPEARANCE:
+                        handle = handle_inbound_appearance
+                    case WS.CAPTCHA:
+                        handle = handle_inbound_captcha
+                    case WS.PONG:
+                        handle = handle_inbound_pong
+                payload = handle(timestamp, queue, user, *parsed)
+
+        # Write to websocket
         if payload is not None:
             queue.put_nowait(payload)