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'''
/remote.php/dav/files/alice/Books/
/remote.php/dav/files/alice/Books/Sci-Fi/
/remote.php/dav/files/alice/Books/Der%20Steppenwolf.epub
4096
/remote.php/dav/files/alice/Books/notes.txt
12
'''
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'''
]>
&lol4;'''
# 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'''
/remote.php/dav/files/alice/Books/
HTTP/1.1 200 OK
/remote.php/dav/files/alice/Books/Sci-Fi/
HTTP/1.1 404 Not Found
HTTP/1.1 200 OK
'''
RELATIVE_HREF_RESPONSE = b'''
Dune.epub
7
HTTP/1.1 200 OK
'''
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')