monkeypatch for local seeking in tubearchivist

dev_range_serve.py

"""local-dev-only HTTP Range support for static/media serving.

`manage.py runserver` (django.views.static.serve) ignores the Range header
entirely, which breaks seeking to a saved video position via the `#t=` media
fragment in the player. Production serves media through nginx, which handles
Range requests natively, so this patch is gated behind DEBUG and never runs
there - see config/settings.py.
"""

import mimetypes
import posixpath
import re
from pathlib import Path

from django.http import FileResponse, Http404, HttpResponse, HttpResponseNotModified
from django.utils._os import safe_join
from django.utils.http import http_date
from django.views.static import was_modified_since

RANGE_RE = re.compile(r"bytes\s*=\s*(\d*)-(\d*)$")


class _BoundedReader:
    """wraps a file object so FileResponse stops reading after `length` bytes"""

    def __init__(self, fileobj, length):
        self._fileobj = fileobj
        self._remaining = length

    def read(self, amount=-1):
        if self._remaining <= 0:
            return b""
        if amount is None or amount < 0 or amount > self._remaining:
            amount = self._remaining
        data = self._fileobj.read(amount)
        self._remaining -= len(data)
        return data

    def close(self):
        self._fileobj.close()


def serve_with_range(request, path, document_root=None, show_indexes=False):
    """drop-in replacement for django.views.static.serve with Range support"""
    path = posixpath.normpath(path).lstrip("/")
    fullpath = Path(safe_join(document_root, path))
    if fullpath.is_dir() or not fullpath.exists():
        raise Http404(f"'{path}' does not exist")

    statobj = fullpath.stat()
    if not was_modified_since(
        request.META.get("HTTP_IF_MODIFIED_SINCE"), statobj.st_mtime
    ):
        return HttpResponseNotModified()

    content_type, encoding = mimetypes.guess_type(str(fullpath))
    content_type = content_type or "application/octet-stream"
    file_size = statobj.st_size

    range_match = RANGE_RE.match(request.META.get("HTTP_RANGE", ""))
    if range_match:
        start_str, end_str = range_match.groups()
        start = int(start_str) if start_str else 0
        end = int(end_str) if end_str else file_size - 1
        end = min(end, file_size - 1)

        if start >= file_size or start > end:
            response = HttpResponse(status=416)
            response["Content-Range"] = f"bytes */{file_size}"
            return response

        length = end - start + 1
        fileobj = fullpath.open("rb")
        fileobj.seek(start)
        response = FileResponse(
            _BoundedReader(fileobj, length),
            status=206,
            content_type=content_type,
        )
        response["Content-Range"] = f"bytes {start}-{end}/{file_size}"
        response["Content-Length"] = str(length)
    else:
        response = FileResponse(fullpath.open("rb"), content_type=content_type)
        response["Content-Length"] = str(file_size)

    response["Accept-Ranges"] = "bytes"
    response["Last-Modified"] = http_date(statobj.st_mtime)
    if encoding:
        response["Content-Encoding"] = encoding

    return response


def patch_static_serve():
    """monkeypatch django.views.static.serve to support Range requests"""
    from django.views import static

    static.serve = serve_with_range

settings.py

"""
Django settings for config project.

Generated by 'django-admin startproject' using Django 3.2.5.

For more information on this file, see
https://docs.djangoproject.com/en/3.2/topics/settings/

For the full list of settings and their values, see
https://docs.djangoproject.com/en/3.2/ref/settings/
"""

import hashlib
from os import environ, path
from pathlib import Path

from common.src.env_settings import EnvironmentSettings
from common.src.helper import ta_host_parser
from common.src.ta_redis import RedisArchivist
from corsheaders.defaults import default_headers

try:
    from dotenv import load_dotenv

    load_dotenv(".env")
except ModuleNotFoundError:
    pass


# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent


# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/3.2/howto/deployment/checklist/

PW_HASH = hashlib.sha256(EnvironmentSettings.TA_PASSWORD.encode())
SECRET_KEY = PW_HASH.hexdigest()

# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = bool(environ.get("DJANGO_DEBUG"))

if DEBUG:
    from common.src.dev_range_serve import patch_static_serve

    patch_static_serve()

ALLOWED_HOSTS, CSRF_TRUSTED_ORIGINS = ta_host_parser(
    environ.get("TA_HOST", "localhost")
)
CORS_ALLOWED_ORIGINS = CSRF_TRUSTED_ORIGINS

# Application definition

INSTALLED_APPS = [
    "django_celery_beat",
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "corsheaders",
    "django.contrib.staticfiles",
    "django.contrib.humanize",
    "rest_framework",
    "rest_framework.authtoken",
    "drf_spectacular",
    "common",
    "video",
    "channel",
    "playlist",
    "download",
    "task",
    "appsettings",
    "stats",
    "user",
    "config",
]

MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    "corsheaders.middleware.CorsMiddleware",
    "config.middleware.StartTimeMiddleware",
    "django.middleware.common.CommonMiddleware",
    "django.middleware.csrf.CsrfViewMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "django.contrib.messages.middleware.MessageMiddleware",
    "django.middleware.clickjacking.XFrameOptionsMiddleware",
]

ROOT_URLCONF = "config.urls"

TEMPLATES = [
    {
        "BACKEND": "django.template.backends.django.DjangoTemplates",
        "DIRS": [],
        "APP_DIRS": True,
        "OPTIONS": {
            "context_processors": [
                "django.template.context_processors.debug",
                "django.template.context_processors.request",
                "django.contrib.auth.context_processors.auth",
                "django.contrib.messages.context_processors.messages",
            ],
        },
    },
]

WSGI_APPLICATION = "config.wsgi.application"

# Database
# https://docs.djangoproject.com/en/3.2/ref/settings/#databases

CACHE_DIR = EnvironmentSettings.CACHE_DIR
DB_PATH = path.join(CACHE_DIR, "db.sqlite3")
DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.sqlite3",
        "NAME": DB_PATH,
    }
}


# Password validation
# https://docs.djangoproject.com/en/3.2/ref/settings/#auth-password-validators

AUTH_PASSWORD_VALIDATORS = [
    {
        "NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator",  # noqa: E501
    },
    {
        "NAME": "django.contrib.auth.password_validation.MinimumLengthValidator",  # noqa: E501
    },
    {
        "NAME": "django.contrib.auth.password_validation.CommonPasswordValidator",  # noqa: E501
    },
    {
        "NAME": "django.contrib.auth.password_validation.NumericPasswordValidator",  # noqa: E501
    },
]

AUTH_USER_MODEL = "user.Account"

# Configure Authentication Backend Combinations
_login_auth_mode = (environ.get("TA_LOGIN_AUTH_MODE") or "single").casefold()
if _login_auth_mode == "local":
    AUTHENTICATION_BACKENDS: tuple = (
        "django.contrib.auth.backends.ModelBackend",
    )
elif _login_auth_mode == "ldap":
    AUTHENTICATION_BACKENDS = ("django_auth_ldap.backend.LDAPBackend",)
    from .ldap_settings import *  # noqa: F403 F401
elif _login_auth_mode == "forwardauth":
    from .fwd_auth_settings import *  # noqa: F403 F401

    AUTHENTICATION_BACKENDS = (
        "django.contrib.auth.backends.RemoteUserBackend",
    )
    MIDDLEWARE.append("user.src.remote_user_auth.HttpRemoteUserMiddleware")
elif _login_auth_mode == "ldap_local":
    AUTHENTICATION_BACKENDS = (
        "django_auth_ldap.backend.LDAPBackend",
        "django.contrib.auth.backends.ModelBackend",
    )
    from .ldap_settings import *  # noqa: F403 F401
else:
    # If none of these cases match, AUTHENTICATION_BACKENDS is unset, which
    # means the ModelBackend should be used by default
    if bool(environ.get("TA_LDAP")):
        AUTHENTICATION_BACKENDS = ("django_auth_ldap.backend.LDAPBackend",)
        from .ldap_settings import *  # noqa: F403 F401
    if bool(environ.get("TA_ENABLE_AUTH_PROXY")):
        from .fwd_auth_settings import *  # noqa: F403 F401

        AUTHENTICATION_BACKENDS = (
            "django.contrib.auth.backends.RemoteUserBackend",
        )
        MIDDLEWARE.append("user.src.remote_user_auth.HttpRemoteUserMiddleware")

# Internationalization
# https://docs.djangoproject.com/en/3.2/topics/i18n/

LANGUAGE_CODE = "en-us"
TIME_ZONE = EnvironmentSettings.TZ
USE_I18N = True
USE_L10N = True
USE_TZ = True


# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/3.2/howto/static-files/

STATIC_URL = "/static/"
STATICFILES_DIRS = (str(BASE_DIR.joinpath("static")),)
STATIC_ROOT = str(BASE_DIR.joinpath("staticfiles"))


# Default primary key field type
# https://docs.djangoproject.com/en/3.2/ref/settings/#default-auto-field

DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"

LOGIN_URL = "/login/"
LOGOUT_REDIRECT_URL = "/login/"

# Cors needed for browser extension
# background.js makes the request so HTTP_ORIGIN will be from extension
if environ.get("DISABLE_CORS"):
    # disable cors
    CORS_ALLOW_ALL_ORIGINS = True
else:
    CORS_ALLOWED_ORIGIN_REGEXES = [
        r"moz-extension://*",
        r"chrome-extension://*",
    ]

CORS_ALLOW_CREDENTIALS = True


CORS_ALLOW_HEADERS = list(default_headers) + [
    "mode", "range"
]
CORS_EXPOSE_HEADERS = ["X-Start-Timestamp", "content-range"]


# TA application settings
TA_UPSTREAM = "https://github.com/tubearchivist/tubearchivist"
TA_VERSION = "v0.5.9"
try:
    TA_START = RedisArchivist().get_message_str("STARTTIMESTAMP")
except ValueError:
    # fails in unittests bootstrap
    pass

# API
REST_FRAMEWORK = {
    "DEFAULT_SCHEMA_CLASS": "drf_spectacular.openapi.AutoSchema",
}

SPECTACULAR_SETTINGS = {
    "TITLE": "Tube Archivist API",
    "DESCRIPTION": "API documentation for Tube Archivist backend.",
    "VERSION": TA_VERSION,
    "SERVE_INCLUDE_SCHEMA": False,
    "SERVE_PERMISSIONS": ["rest_framework.permissions.IsAuthenticated"],
}

# Logging configuration
LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
        },
    },
    "loggers": {
        "apprise": {
            "handlers": ["console"],
            "level": "DEBUG",
            "propagate": True,
        },
    },
}