diora-web/accounts/tests.py

import json
import socket
from types import SimpleNamespace
from unittest.mock import patch

from django.contrib.auth.models import User
from django.test import TestCase, override_settings

from .models import ApiToken, WebDAVSource
from .webdav import WebDAVError, assert_safe_url, fetch_file, list_directory, safe_rel_path
from books.models import EBook, EBookProgress, EBookHighlights


class ApiTokenAuthMiddlewareTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')
        self.token = ApiToken.objects.create(user=self.user)

    def test_valid_bearer_token_authenticates(self):
        resp = self.client.get('/books/', HTTP_AUTHORIZATION=f'Bearer {self.token.token}')
        self.assertEqual(resp.status_code, 200)

    def test_missing_token_is_unauthenticated(self):
        resp = self.client.get('/books/')
        self.assertEqual(resp.status_code, 401)

    def test_invalid_token_is_unauthenticated(self):
        resp = self.client.get('/books/', HTTP_AUTHORIZATION='Bearer not-a-real-token')
        self.assertEqual(resp.status_code, 401)

    def test_token_scoped_to_owning_user(self):
        other = User.objects.create_user(username='bob', password='pw12345678')
        EBook.objects.create(user=other, meta_ct='ct', meta_iv='iv', data_ct='ct', data_iv='iv')
        resp = self.client.get('/books/', HTTP_AUTHORIZATION=f'Bearer {self.token.token}')
        self.assertEqual(resp.json(), [])

    def test_regenerate_invalidates_old_token(self):
        old = self.token.token
        self.token.regenerate()
        resp = self.client.get('/books/', HTTP_AUTHORIZATION=f'Bearer {old}')
        self.assertEqual(resp.status_code, 401)
        resp = self.client.get('/books/', HTTP_AUTHORIZATION=f'Bearer {self.token.token}')
        self.assertEqual(resp.status_code, 200)

    def test_regenerate_view_requires_login_and_rotates_token(self):
        resp = self.client.post('/accounts/api-token/regenerate/')
        self.assertEqual(resp.status_code, 302)
        self.assertIn('/accounts/login/', resp.url)

        self.client.login(username='alice', password='pw12345678')
        old = self.token.token
        resp = self.client.post('/accounts/api-token/regenerate/')
        self.assertRedirects(resp, '/accounts/settings/')
        self.token.refresh_from_db()
        self.assertNotEqual(self.token.token, old)


class SyncSnapshotTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')
        self.token = ApiToken.objects.create(user=self.user)
        self.book = EBook.objects.create(
            user=self.user, meta_ct='meta-ct', meta_iv='meta-iv', data_ct='data-ct', data_iv='data-iv',
        )
        EBookProgress.objects.create(
            user=self.user, book=self.book, scroll_fraction=0.5, position_anchor='3:0.5',
        )
        EBookHighlights.objects.create(user=self.user, book=self.book, ct='hl-ct', iv='hl-iv')

    def test_requires_auth(self):
        resp = self.client.get('/api/sync/')
        self.assertEqual(resp.status_code, 401)

    def test_snapshot_shape_and_data(self):
        resp = self.client.get('/api/sync/', HTTP_AUTHORIZATION=f'Bearer {self.token.token}')
        self.assertEqual(resp.status_code, 200)
        data = resp.json()
        for key in (
            'server_time', 'books', 'book_progress', 'book_highlights', 'book_bookmarks',
            'podcast_feeds', 'episode_progress', 'podcast_queue', 'saved_stations',
        ):
            self.assertIn(key, data)

        self.assertEqual(len(data['books']), 1)
        self.assertEqual(data['books'][0]['meta_ct'], 'meta-ct')

        self.assertEqual(len(data['book_progress']), 1)
        self.assertEqual(data['book_progress'][0]['book_id'], self.book.id)
        self.assertEqual(data['book_progress'][0]['position_anchor'], '3:0.5')

        self.assertEqual(len(data['book_highlights']), 1)
        self.assertEqual(data['book_highlights'][0]['ct'], 'hl-ct')

        # No data_ct/data_iv leaked into the snapshot (book bytes stay lazy-fetched)
        self.assertNotIn('data_ct', data['books'][0])


# ---------------------------------------------------------------------------
# WebDAV cloud import
# ---------------------------------------------------------------------------

PROPFIND_RESPONSE = b'''<?xml version="1.0"?>
<d:multistatus xmlns:d="DAV:">
  <d:response>
    <d:href>/remote.php/dav/files/alice/Books/</d:href>
    <d:propstat><d:prop><d:resourcetype><d:collection/></d:resourcetype></d:prop></d:propstat>
  </d:response>
  <d:response>
    <d:href>/remote.php/dav/files/alice/Books/Sci-Fi/</d:href>
    <d:propstat><d:prop><d:resourcetype><d:collection/></d:resourcetype></d:prop></d:propstat>
  </d:response>
  <d:response>
    <d:href>/remote.php/dav/files/alice/Books/Der%20Steppenwolf.epub</d:href>
    <d:propstat><d:prop><d:resourcetype/><d:getcontentlength>4096</d:getcontentlength></d:prop></d:propstat>
  </d:response>
  <d:response>
    <d:href>/remote.php/dav/files/alice/Books/notes.txt</d:href>
    <d:propstat><d:prop><d:resourcetype/><d:getcontentlength>12</d:getcontentlength></d:prop></d:propstat>
  </d:response>
</d:multistatus>'''


class _FakeSocket:
    def __init__(self, peer):
        self._peer = peer

    def getpeername(self):
        return (self._peer, 443)


class _FakeResponse:
    def __init__(self, status_code=207, content=b'', headers=None, peer=None):
        self.status_code = status_code
        self.content = content
        self.headers = headers or {}
        self.closed = False
        # Mirrors requests' response.raw._connection.sock, which is what
        # _assert_peer_is_safe introspects. None means "nothing to check".
        if peer is None:
            self.raw = None
        else:
            self.raw = SimpleNamespace(_connection=SimpleNamespace(sock=_FakeSocket(peer)))

    def iter_content(self, chunk_size=None):
        yield self.content

    def close(self):
        self.closed = True


class NormalizedBaseUrlTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')

    def _source(self, **kwargs):
        kwargs.setdefault('label', 'cloud')
        kwargs.setdefault('username', 'alice')
        return WebDAVSource(user=self.user, **kwargs)

    def test_bare_host_expands_to_nextcloud_files_endpoint(self):
        source = self._source(base_url='https://cloud.example.com')
        self.assertEqual(
            source.normalized_base_url(),
            'https://cloud.example.com/remote.php/dav/files/alice/',
        )

    def test_root_path_is_appended(self):
        source = self._source(base_url='https://cloud.example.com', root_path='/Buecher/')
        self.assertEqual(
            source.normalized_base_url(),
            'https://cloud.example.com/remote.php/dav/files/alice/Buecher/',
        )

    def test_explicit_dav_url_is_left_alone(self):
        source = self._source(base_url='https://dav.example.com/webdav')
        self.assertEqual(source.normalized_base_url(), 'https://dav.example.com/webdav/')

    def test_generic_server_without_username_is_not_rewritten(self):
        source = self._source(base_url='https://files.example.com/share', username='')
        self.assertEqual(source.normalized_base_url(), 'https://files.example.com/share/')


class SafeRelPathTests(TestCase):
    def test_traversal_is_rejected(self):
        with self.assertRaises(WebDAVError):
            safe_rel_path('Books/../../etc/passwd')

    def test_leading_slashes_and_dots_are_stripped(self):
        self.assertEqual(safe_rel_path('/Books/./Sci-Fi/'), 'Books/Sci-Fi')

    def test_empty_path_is_root(self):
        self.assertEqual(safe_rel_path(''), '')
        self.assertEqual(safe_rel_path('/'), '')


class AssertSafeUrlTests(TestCase):
    def _resolve_to(self, ip):
        return [(2, 1, 6, '', (ip, 443))]

    def test_non_http_scheme_rejected(self):
        with self.assertRaises(WebDAVError):
            assert_safe_url('file:///etc/passwd')

    def test_private_address_rejected(self):
        with patch('accounts.webdav.socket.getaddrinfo', return_value=self._resolve_to('172.18.0.4')):
            with self.assertRaises(WebDAVError):
                assert_safe_url('https://internal.example.com/dav/')

    def test_loopback_rejected(self):
        with patch('accounts.webdav.socket.getaddrinfo', return_value=self._resolve_to('127.0.0.1')):
            with self.assertRaises(WebDAVError):
                assert_safe_url('http://localhost:11000/remote.php/dav/')

    def test_link_local_metadata_endpoint_rejected(self):
        with patch('accounts.webdav.socket.getaddrinfo', return_value=self._resolve_to('169.254.169.254')):
            with self.assertRaises(WebDAVError):
                assert_safe_url('http://metadata.example.com/')

    def test_public_address_allowed(self):
        with patch('accounts.webdav.socket.getaddrinfo', return_value=self._resolve_to('85.214.6.118')):
            assert_safe_url('https://nc.example.com/remote.php/dav/')

    def test_unresolvable_host_rejected(self):
        with patch('accounts.webdav.socket.getaddrinfo', side_effect=socket.gaierror):
            with self.assertRaises(WebDAVError):
                assert_safe_url('https://nope.example.com/')

    @override_settings(WEBDAV_ALLOW_PRIVATE_HOSTS=True)
    def test_private_allowed_when_opted_in(self):
        assert_safe_url('http://192.168.1.10/dav/')


class ListDirectoryTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')
        self.source = WebDAVSource.objects.create(
            user=self.user, label='cloud', base_url='https://cloud.example.com',
            username='alice', password='app-pw', root_path='Books',
        )

    def _list(self, response=None):
        with patch('accounts.webdav.assert_safe_url'), \
             patch('accounts.webdav.requests.request',
                   return_value=response or _FakeResponse(content=PROPFIND_RESPONSE)):
            return list_directory(self.source)

    def test_entries_are_parsed_and_sorted_dirs_first(self):
        entries = self._list()
        self.assertEqual([e['name'] for e in entries],
                         ['Sci-Fi', 'Der Steppenwolf.epub', 'notes.txt'])

    def test_collection_itself_is_excluded(self):
        self.assertNotIn('Books', [e['name'] for e in self._list()])

    def test_book_flag_and_size(self):
        by_name = {e['name']: e for e in self._list()}
        self.assertTrue(by_name['Der Steppenwolf.epub']['is_book'])
        self.assertEqual(by_name['Der Steppenwolf.epub']['size'], 4096)
        self.assertFalse(by_name['notes.txt']['is_book'])
        self.assertTrue(by_name['Sci-Fi']['is_dir'])

    def test_redirect_is_refused_rather_than_followed(self):
        redirect_response = _FakeResponse(status_code=302, headers={'Location': 'http://127.0.0.1/'})
        with self.assertRaises(WebDAVError):
            self._list(redirect_response)

    def test_bad_credentials_surface_clearly(self):
        with self.assertRaises(WebDAVError) as ctx:
            self._list(_FakeResponse(status_code=401))
        self.assertIn('App-Passwort', str(ctx.exception))


class FetchFileTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')
        self.source = WebDAVSource.objects.create(
            user=self.user, label='cloud', base_url='https://cloud.example.com', username='alice',
        )

    def test_non_book_extension_refused_before_any_request(self):
        with patch('accounts.webdav.requests.request') as mock_request:
            with self.assertRaises(WebDAVError):
                fetch_file(self.source, 'secrets.env', 1024)
        mock_request.assert_not_called()

    def test_declared_oversize_refused(self):
        response = _FakeResponse(status_code=200, headers={'Content-Length': '99999'})
        with patch('accounts.webdav.assert_safe_url'), \
             patch('accounts.webdav.requests.request', return_value=response):
            with self.assertRaises(WebDAVError):
                fetch_file(self.source, 'big.epub', 1024)

    def test_streamed_oversize_refused_even_without_content_length(self):
        response = _FakeResponse(status_code=200, content=b'x' * 5000)
        with patch('accounts.webdav.assert_safe_url'), \
             patch('accounts.webdav.requests.request', return_value=response):
            with self.assertRaises(WebDAVError):
                fetch_file(self.source, 'sneaky.epub', 1024)

    def test_successful_fetch_returns_bytes(self):
        response = _FakeResponse(status_code=200, content=b'EPUB-BYTES')
        with patch('accounts.webdav.assert_safe_url'), \
             patch('accounts.webdav.requests.request', return_value=response):
            self.assertEqual(fetch_file(self.source, 'ok.epub', 1024), b'EPUB-BYTES')


class CloudImportViewTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')
        self.other = User.objects.create_user(username='bob', password='pw12345678')
        self.source = WebDAVSource.objects.create(
            user=self.user, label='cloud', base_url='https://cloud.example.com', username='alice',
        )
        self.foreign = WebDAVSource.objects.create(
            user=self.other, label='bobs', base_url='https://other.example.com', username='bob',
        )

    def test_endpoints_require_authentication(self):
        self.assertEqual(self.client.get('/books/cloud/sources/').status_code, 401)
        self.assertEqual(self.client.get(f'/books/cloud/{self.source.pk}/browse/').status_code, 401)
        self.assertEqual(self.client.get(f'/books/cloud/{self.source.pk}/fetch/?path=a.epub').status_code, 401)

    def test_sources_are_scoped_to_the_owner(self):
        self.client.force_login(self.user)
        labels = [s['label'] for s in self.client.get('/books/cloud/sources/').json()['sources']]
        self.assertEqual(labels, ['cloud'])

    def test_foreign_source_is_not_browsable(self):
        self.client.force_login(self.user)
        resp = self.client.get(f'/books/cloud/{self.foreign.pk}/browse/')
        self.assertEqual(resp.status_code, 404)

    def test_browse_returns_entries(self):
        self.client.force_login(self.user)
        entries = [{'name': 'Dune.epub', 'path': 'Dune.epub', 'is_dir': False,
                    'size': 10, 'modified': '', 'is_book': True}]
        with patch('books.webdav.list_directory', return_value=entries):
            resp = self.client.get(f'/books/cloud/{self.source.pk}/browse/')
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(resp.json()['entries'], entries)

    def test_browse_reports_upstream_failure_as_502(self):
        self.client.force_login(self.user)
        with patch('books.webdav.list_directory', side_effect=WebDAVError('kaputt')):
            resp = self.client.get(f'/books/cloud/{self.source.pk}/browse/')
        self.assertEqual(resp.status_code, 502)
        self.assertEqual(resp.json()['error'], 'kaputt')

    def test_bad_user_input_is_400_not_502(self):
        self.client.force_login(self.user)
        resp = self.client.get(f'/books/cloud/{self.source.pk}/fetch/?path=../../etc/passwd')
        self.assertEqual(resp.status_code, 400)
        resp = self.client.get(f'/books/cloud/{self.source.pk}/fetch/?path=notes.txt')
        self.assertEqual(resp.status_code, 400)

    def test_fetch_streams_bytes_without_storing_them(self):
        self.client.force_login(self.user)
        with patch('books.webdav.fetch_file', return_value=b'EPUB-BYTES'):
            resp = self.client.get(f'/books/cloud/{self.source.pk}/fetch/?path=Dune.epub')
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(resp.content, b'EPUB-BYTES')
        # The proxy is a pass-through: nothing is persisted server-side.
        self.assertEqual(EBook.objects.count(), 0)


BILLION_LAUGHS = b'''<?xml version="1.0"?>
<!DOCTYPE lolz [
 <!ENTITY lol "lol">
 <!ENTITY lol1 "&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;">
 <!ENTITY lol2 "&lol1;&lol1;&lol1;&lol1;&lol1;&lol1;&lol1;&lol1;&lol1;&lol1;">
 <!ENTITY lol3 "&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;">
 <!ENTITY lol4 "&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;">
]>
<d:multistatus xmlns:d="DAV:"><d:response><d:href>&lol4;</d:href></d:response></d:multistatus>'''

# A server that answers with the 404 propstat first — legal per RFC 4918, and
# what made directories disappear from the listing before _select_prop existed.
PROPSTAT_404_FIRST = b'''<?xml version="1.0"?>
<d:multistatus xmlns:d="DAV:">
  <d:response>
    <d:href>/remote.php/dav/files/alice/Books/</d:href>
    <d:propstat><d:prop><d:resourcetype><d:collection/></d:resourcetype></d:prop>
      <d:status>HTTP/1.1 200 OK</d:status></d:propstat>
  </d:response>
  <d:response>
    <d:href>/remote.php/dav/files/alice/Books/Sci-Fi/</d:href>
    <d:propstat><d:prop><d:getcontentlength/></d:prop>
      <d:status>HTTP/1.1 404 Not Found</d:status></d:propstat>
    <d:propstat><d:prop><d:resourcetype><d:collection/></d:resourcetype></d:prop>
      <d:status>HTTP/1.1 200 OK</d:status></d:propstat>
  </d:response>
</d:multistatus>'''

RELATIVE_HREF_RESPONSE = b'''<?xml version="1.0"?>
<d:multistatus xmlns:d="DAV:">
  <d:response><d:href>Dune.epub</d:href>
    <d:propstat><d:prop><d:resourcetype/><d:getcontentlength>7</d:getcontentlength></d:prop>
      <d:status>HTTP/1.1 200 OK</d:status></d:propstat>
  </d:response>
</d:multistatus>'''


class WebDAVHardeningTests(TestCase):
    """Regressions for the SSRF / resource-exhaustion review findings."""

    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')
        self.source = WebDAVSource.objects.create(
            user=self.user, label='cloud', base_url='https://cloud.example.com',
            username='alice', password='app-pw', root_path='Books',
        )

    def _list(self, response):
        with patch('accounts.webdav.assert_safe_url'), \
             patch('accounts.webdav.requests.request', return_value=response):
            return list_directory(self.source)

    # --- DNS rebinding -----------------------------------------------------

    def test_peer_address_is_rechecked_after_connecting(self):
        """A resolver that answers public-then-private must not leak a body.

        assert_safe_url passes (it is given a public answer), but the socket
        actually landed on a Docker-internal address.
        """
        response = _FakeResponse(content=PROPFIND_RESPONSE, peer='172.18.0.5')
        with self.assertRaises(WebDAVError):
            self._list(response)
        self.assertTrue(response.closed)

    def test_public_peer_is_accepted(self):
        response = _FakeResponse(content=PROPFIND_RESPONSE, peer='85.214.6.118')
        self.assertTrue(self._list(response))

    @override_settings(WEBDAV_ALLOW_PRIVATE_HOSTS=True)
    def test_peer_check_respects_the_opt_out(self):
        response = _FakeResponse(content=PROPFIND_RESPONSE, peer='192.168.1.10')
        self.assertTrue(self._list(response))

    # --- Resource exhaustion ----------------------------------------------

    def test_entity_expansion_is_refused(self):
        with self.assertRaises(WebDAVError):
            self._list(_FakeResponse(content=BILLION_LAUGHS))

    def test_oversized_listing_is_refused(self):
        oversized = _FakeResponse(headers={'Content-Length': str(9 * 1024 * 1024)})
        with self.assertRaises(WebDAVError):
            self._list(oversized)
        self.assertTrue(oversized.closed)

    # --- Information disclosure -------------------------------------------

    def test_rejection_message_leaks_neither_ip_nor_resolvability(self):
        private = [(2, 1, 6, '', ('172.18.0.5', 443))]
        with patch('accounts.webdav.socket.getaddrinfo', return_value=private):
            with self.assertRaises(WebDAVError) as private_ctx:
                assert_safe_url('http://forgejo-db/')
        with patch('accounts.webdav.socket.getaddrinfo', side_effect=socket.gaierror):
            with self.assertRaises(WebDAVError) as missing_ctx:
                assert_safe_url('http://forgejo-db/')

        self.assertNotIn('172.18.0.5', str(private_ctx.exception))
        # Same wording either way, so the endpoint cannot be used to tell an
        # existing internal host from a nonexistent one.
        self.assertEqual(str(private_ctx.exception), str(missing_ctx.exception))

    # --- Malformed input ---------------------------------------------------

    def test_invalid_port_is_reported_not_crashed(self):
        with self.assertRaises(WebDAVError):
            assert_safe_url('https://example.com:99999/dav/')

    def test_ipv6_transition_ranges_are_rejected(self):
        for address in ('64:ff9b::7f00:1', '::127.0.0.1'):
            with patch('accounts.webdav.socket.getaddrinfo',
                       return_value=[(10, 1, 6, '', (address, 443, 0, 0))]):
                with self.assertRaises(WebDAVError, msg=address):
                    assert_safe_url('https://nat64.example.com/')

    # --- PROPFIND parsing --------------------------------------------------

    def test_directory_survives_a_404_propstat_listed_first(self):
        entries = self._list(_FakeResponse(content=PROPSTAT_404_FIRST))
        by_name = {e['name']: e for e in entries}
        self.assertTrue(by_name['Sci-Fi']['is_dir'])

    def test_relative_hrefs_are_resolved(self):
        entries = self._list(_FakeResponse(content=RELATIVE_HREF_RESPONSE))
        self.assertEqual([e['name'] for e in entries], ['Dune.epub'])

    # --- URL normalisation -------------------------------------------------

    def test_host_named_webdav_still_gets_the_nextcloud_path(self):
        source = WebDAVSource(user=self.user, label='x',
                              base_url='https://webdav.example.com', username='alice')
        self.assertEqual(
            source.normalized_base_url(),
            'https://webdav.example.com/remote.php/dav/files/alice/',
        )


class WebDAVSourceLimitTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')
        self.client.force_login(self.user)

    @override_settings(WEBDAV_MAX_SOURCES_PER_USER=2)
    def test_sources_are_capped_per_user(self):
        with patch('accounts.views._probe_source', return_value=(20, 'ok')):
            for i in range(3):
                self.client.post('/accounts/webdav/add/',
                                 {'label': f'c{i}', 'base_url': 'https://example.com'})
        self.assertEqual(self.user.webdav_sources.count(), 2)


class CloudFetchCachingTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')
        self.source = WebDAVSource.objects.create(
            user=self.user, label='cloud', base_url='https://cloud.example.com', username='alice',
        )

    def test_plaintext_bytes_are_not_cacheable(self):
        self.client.force_login(self.user)
        with patch('books.webdav.fetch_file', return_value=b'EPUB-BYTES'):
            resp = self.client.get(f'/books/cloud/{self.source.pk}/fetch/?path=Dune.epub')
        self.assertEqual(resp['Cache-Control'], 'no-store')