Access captcha

anonstream/__init__.py

@@ -38,6 +38,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 = {}
+
     # State for tasks
     app.users_update_buffer = set()
     app.stream_title = None
@@ -77,7 +79,6 @@ def create_app(toml_config):
                 )
             app.add_background_task(start_event_server)
 
-
         # Create routes and background tasks
         import anonstream.routes
         import anonstream.tasks

anonstream/access.py

@@ -0,0 +1,18 @@
+import time
+
+from quart import current_app
+
+FAILURES = current_app.failures
+
+def add_failure(message):
+    timestamp = time.time_ns() // 1_000_000
+    while timestamp in FAILURES:
+        timestamp += 1
+    FAILURES[timestamp] = message
+    return timestamp
+
+def pop_failure(failure_id):
+    try:
+        return FAILURES.pop(failure_id)
+    except KeyError:
+        return None

anonstream/config.py

@@ -19,6 +19,7 @@ def update_flask_from_toml(toml_config, flask_config):
         'AUTH_USERNAME': toml_config['auth']['username'],
         'AUTH_PWHASH': auth_pwhash,
         'AUTH_TOKEN': generate_token(),
+        'ACCESS_CAPTCHA': toml_config['access']['captcha'],
      })
     for flask_section in toml_to_flask_sections(toml_config):
         flask_config.update(flask_section)

anonstream/routes/core.py

@@ -6,23 +6,38 @@ import math
 from quart import current_app, request, render_template, abort, make_response, redirect, url_for, abort, send_from_directory
 from werkzeug.exceptions import TooManyRequests
 
-from anonstream.captcha import get_captcha_image
+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
-from anonstream.routes.wrappers import with_user_from, auth_required, clean_cache_headers
+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
 
+CAPTCHA_SIGNER = current_app.captcha_signer
 STATIC_DIRECTORY = current_app.root_path / 'static'
 
 @current_app.route('/')
-@with_user_from(request)
-async def home(timestamp, user):
-    return await render_template(
-        'home.html',
-        csp=generate_csp(),
-        user=user,
-    )
+@with_user_from(request, fallback_to_token=True)
+async def home(timestamp, user_or_token):
+    match user_or_token:
+        case str() | None:
+            failure_id = request.args.get('failure', type=int)
+            response = await render_template(
+                'captcha.html',
+                csp=generate_csp(),
+                token=user_or_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,
+            )
+    return response
 
 @current_app.route('/stream.mp4')
 @with_user_from(request)
@@ -58,8 +73,8 @@ async def login():
     return redirect(url_for('home'), 303)
 
 @current_app.route('/captcha.jpg')
-@with_user_from(request)
-async def captcha(timestamp, user):
+@with_user_from(request, fallback_to_token=True)
+async def captcha(timestamp, user_or_token):
     digest = request.args.get('digest', '')
     image = get_captcha_image(digest)
     if image is None:
@@ -67,6 +82,33 @@ async def captcha(timestamp, user):
     else:
         return image, {'Content-Type': 'image/jpeg'}
 
+@current_app.post('/access')
+@with_user_from(request, fallback_to_token=True)
+async def access(timestamp, user_or_token):
+    match user_or_token:
+        case str() | None:
+            token = user_or_token
+            form = await request.form
+            digest = form.get('digest', '')
+            answer = form.get('answer', '')
+            match check_captcha_digest(CAPTCHA_SIGNER, digest, answer):
+                case Answer.MISSING:
+                    failure_id = add_failure('Captcha is required')
+                case Answer.BAD:
+                    failure_id = add_failure('Captcha was incorrect')
+                case Answer.EXPIRED:
+                    failure_id = add_failure('Captcha has expired')
+                case Answer.OK:
+                    failure_id = None
+                    user = generate_and_add_user(timestamp, 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
+    url = url_for('home', token=user['token'])
+    return redirect(url, 303)
+
 @current_app.route('/static/<filename>')
 @with_user_from(request)
 @clean_cache_headers

anonstream/routes/wrappers.py

@@ -71,50 +71,69 @@ def auth_required(f):
 
     return wrapper
 
-def with_user_from(context):
+def generate_and_add_user(timestamp, token=None, broadcaster=False):
+    token = token or generate_token()
+    user = generate_user(
+        timestamp=timestamp,
+        token=token,
+        broadcaster=broadcaster,
+        presence=Presence.NOTWATCHING,
+    )
+    USERS_BY_TOKEN[token] = user
+    USERS_UPDATE_BUFFER.add(token)
+    return user
+
+def with_user_from(context, fallback_to_token=False):
     def with_user_from_context(f):
         @wraps(f)
         async def wrapper(*args, **kwargs):
             timestamp = get_timestamp()
 
-            # Check if broadcaster
+            # Get token
             broadcaster = check_auth(context)
+            token_from_args = context.args.get('token')
+            token_from_cookie = try_unquote(context.cookies.get('token'))
+            token_from_context = token_from_args or token_from_cookie
             if broadcaster:
                 token = CONFIG['AUTH_TOKEN']
+            elif CONFIG['ACCESS_CAPTCHA']:
+                token = token_from_context
             else:
-                token = (
-                    context.args.get('token')
-                    or try_unquote(context.cookies.get('token'))
-                    or generate_token()
-                )
-                if hmac.compare_digest(token, CONFIG['AUTH_TOKEN']):
-                    raise abort(401)
+                token = token_from_context or generate_token()
 
             # Reject invalid tokens
-            if not RE_TOKEN.fullmatch(token):
+            if isinstance(token, str) and not RE_TOKEN.fullmatch(token):
                 raise abort(400)
 
-            # Update / create user
-            user = USERS_BY_TOKEN.get(token)
-            if user is not None:
-                see(user)
-            else:
-                user = generate_user(
-                    timestamp=timestamp,
-                    token=token,
-                    broadcaster=broadcaster,
-                    presence=Presence.NOTWATCHING,
-                )
-                USERS_BY_TOKEN[token] = user
+            # Only logged in broadcaster may have the broadcaster's token
+            if (
+                not broadcaster
+                and isinstance(token, str)
+                and hmac.compare_digest(token, CONFIG['AUTH_TOKEN'])
+            ):
+                    raise abort(401)
 
-                # Add to the users update buffer
-                USERS_UPDATE_BUFFER.add(token)
+            # Create response
+            user = USERS_BY_TOKEN.get(token)
+            if CONFIG['ACCESS_CAPTCHA'] and not broadcaster:
+                if user is not None:
+                    user['last']['seen'] = timestamp
+                    response = await f(timestamp, user, *args, **kwargs)
+                elif fallback_to_token:
+                    #assert not broadcaster
+                    response = await f(timestamp, token, *args, **kwargs)
+                else:
+                    raise abort(403)
+            else:
+                if user is None:
+                    user = generate_and_add_user(timestamp, token, broadcaster)
+                response = await f(timestamp, user, *args, **kwargs)
 
             # Set cookie
-            response = await f(timestamp, user, *args, **kwargs)
-            if try_unquote(context.cookies.get('token')) != token:
+            if token_from_cookie != token:
                 response = await make_response(response)
                 response.headers['Set-Cookie'] = f'token={quote(token)}; path=/'
+
             return response
 
         return wrapper

anonstream/templates/captcha.html

@@ -0,0 +1,60 @@
+<!doctype html>
+<html>
+<head>
+  <meta charset="utf-8">
+  <meta http-equiv="content-security-policy" content="default-src 'none'; img-src 'self'; style-src 'nonce-{{ csp }}';">
+  <style nonce="{{ csp }}">
+    body {
+      background-color: #232327;
+      color: #ddd;
+      font-family: sans-serif;
+      font-size: 14pt;
+      display: grid;
+      grid-template-rows: calc(50% - 4rem) 1fr;
+      height: 100vh;
+      margin: 0;
+      padding: 1rem;
+      box-sizing: border-box;
+    }
+    img {
+      margin: auto auto 1rem;
+      border: 1px solid #222;
+    }
+    form {
+      margin: 0 auto auto;
+      display: grid;
+      grid-template-columns: auto auto;
+      grid-gap: 0.5rem;
+    }
+    input[name="answer"] {
+      background-color: #47474a;
+      border: 1px solid #777;
+      border-radius: 2px;
+      color: #ddd;
+      font-size: 14pt;
+      padding: 1px 3px;
+      width: 10ch;
+    }
+    input[name="answer"]:hover {
+      background-color: #37373a;
+      transition: 0.25s;
+	}
+	input[type="submit"] {
+      font-size: 14pt;
+	}
+	p {
+      grid-column: 1 / span 2;
+      text-align: center;
+	}
+  </style>
+</head>
+<body>
+  <img src="{{ url_for('captcha', digest=digest) }}" width="72" height="30">
+  <form action="{{ url_for('access', token=token) }}" method="post">
+    <input type="hidden" name="digest" value="{{ digest }}">
+    <input name="answer" placeholder="Captcha" required autofocus>
+    <input type="submit" value="Submit">
+    {% if failure is not none %}<p>{{ failure }}</p>{% endif %}
+  </form>
+</body>
+</html>

config.toml

@@ -24,6 +24,9 @@ stream_initial_buffer = 3
 file = "title.txt"
 file_cache_lifetime = 0.5
 
+[access]
+captcha = true
+
 [captcha]
 lifetime = 1800
 fonts = []