diora-web/feeds/logins.py

"""Subscriber logins for sites behind a paywall (junge Welt …).

The account holder enters the site's username and password in the feed's
settings; the server logs in with them and fetches the articles as a
subscriber. The password is stored encrypted (Fernet, key derived from
SECRET_KEY — a copy of the database alone doesn't reveal it) and never
handed out again. Each supported site is an adapter: how to log in, how a
paywalled page looks.
"""

import base64
import hashlib
import logging
import threading
import time
from urllib.parse import urlparse

import requests
from cryptography.fernet import Fernet, InvalidToken
from django.conf import settings
from django.utils import timezone

log = logging.getLogger(__name__)

TIMEOUT = 15
SESSION_SECONDS = 6 * 3600   # a logged-in session is reused this long, then logs in again


class LoginFailed(Exception):
    pass


class JungeWelt:
    domain = 'jungewelt.de'
    name = 'junge Welt'

    def login(self, http, username, password):
        r = http.post('https://www.jungewelt.de/login.php', timeout=TIMEOUT, allow_redirects=False,
                      data={'username': username, 'password': password, 'ref': '/', 'submit': 'Login'})
        where = r.headers.get('Location', '')
        if 'loginFailed' in where:
            raise LoginFailed('Benutzername oder Passwort falsch')
        if r.status_code not in (301, 302, 303) or not http.cookies.get('jW'):
            raise LoginFailed(f'Anmeldung fehlgeschlagen ({r.status_code})')

    def paywalled(self, page):
        return 'mit einem Abonnement und erhalten Sie Zugang' in page


ADAPTERS = {a.domain: a for a in (JungeWelt(),)}


def adapter_for(url):
    host = (urlparse(url or '').hostname or '').lower()
    for domain, adapter in ADAPTERS.items():
        if host == domain or host.endswith('.' + domain):
            return adapter
    return None


def _fernet():
    key = hashlib.sha256(b'diora feeds site login\0' + settings.SECRET_KEY.encode()).digest()
    return Fernet(base64.urlsafe_b64encode(key))


def encrypt(text):
    return _fernet().encrypt(text.encode()).decode()


def decrypt(token):
    """The plain text, or None when it can't be read (SECRET_KEY changed)."""
    try:
        return _fernet().decrypt(token.encode()).decode()
    except (InvalidToken, ValueError):
        return None


def new_http():
    from .fetch import USER_AGENT
    s = requests.Session()
    s.headers['User-Agent'] = USER_AGENT
    return s


# Logged-in sessions in this process: (user id, domain) -> (session, logged in at).
_sessions = {}
_lock = threading.Lock()


def forget(user_id, domain):
    with _lock:
        _sessions.pop((user_id, domain), None)


def check_and_store(user, url, username, password):
    """Log in once; on success store the login (encrypted). Raises LoginFailed."""
    from .models import SiteLogin
    adapter = adapter_for(url)
    if adapter is None:
        raise LoginFailed('Für diese Seite gibt es keine Anmeldung.')
    http = new_http()
    try:
        adapter.login(http, username, password)
    except requests.RequestException:
        raise LoginFailed(f'{adapter.name} ist nicht erreichbar.')
    SiteLogin.objects.update_or_create(user=user, domain=adapter.domain, defaults={
        'username': username, 'secret': encrypt(password), 'ok': True, 'last_error': '',
        'checked_at': timezone.now()})
    with _lock:
        _sessions[(user.id, adapter.domain)] = (http, time.monotonic())
    return adapter


def session_for(user_id, url, fresh=False):
    """A logged-in requests session for the article's site, or None (no login
    stored, or it no longer works — then the login is marked as failing)."""
    from .models import SiteLogin
    adapter = adapter_for(url)
    if adapter is None:
        return None
    key = (user_id, adapter.domain)
    with _lock:
        cached = _sessions.get(key)
        if cached and not fresh and time.monotonic() - cached[1] < SESSION_SECONDS:
            return cached[0]
    login = SiteLogin.objects.filter(user_id=user_id, domain=adapter.domain).first()
    if login is None:
        return None
    password = decrypt(login.secret)
    if password is None:
        login.ok, login.last_error = False, 'Bitte die Zugangsdaten neu eingeben.'
        login.save(update_fields=['ok', 'last_error'])
        return None
    http = new_http()
    try:
        adapter.login(http, login.username, password)
    except (LoginFailed, requests.RequestException) as e:
        login.ok = False
        login.last_error = str(e) if isinstance(e, LoginFailed) else f'{adapter.name} nicht erreichbar'
        login.checked_at = timezone.now()
        login.save(update_fields=['ok', 'last_error', 'checked_at'])
        log.info('site login %s for user %s failed: %s', adapter.domain, user_id, e)
        return None
    if not login.ok:
        login.ok, login.last_error = True, ''
        login.save(update_fields=['ok', 'last_error'])
    with _lock:
        _sessions[key] = (http, time.monotonic())
    return http


def has_login(user_id, url):
    from .models import SiteLogin
    adapter = adapter_for(url)
    return bool(adapter) and SiteLogin.objects.filter(user_id=user_id, domain=adapter.domain, ok=True).exists()


def info(user_id, url):
    """What the feed settings show about the login for this source's site."""
    from .models import SiteLogin
    adapter = adapter_for(url)
    if adapter is None:
        return None
    login = SiteLogin.objects.filter(user_id=user_id, domain=adapter.domain).first()
    return {'site': adapter.name, 'domain': adapter.domain,
            'username': login.username if login else '', 'ok': bool(login and login.ok),
            'error': login.last_error if login else ''}