mrwnslz
69442f4a4d
TranscriptionSettings (Token Fernet-verschluesselt wie SiteLogin), TranscriptionJob und EpisodeArticle in podcasts; Source.synthetic in feeds, das Refresh, Aufraeumen, OPML-Export, Kommando und Umstellen auf Nachrichten ueberspringt bzw. verweigert. GET/POST /podcasts/transcription/settings/ und POST /podcasts/transcription/test/ (SSRF-Pruefung ueber chapters._open, jetzt auch fuer POST; eingebaute Instanz aus DIORA_WHISPER_URL ausgenommen). Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
153 lines
6.9 KiB
Python
153 lines
6.9 KiB
Python
import io
|
|
import json
|
|
import wave
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.test import TestCase, override_settings
|
|
|
|
from feeds import logins
|
|
|
|
from . import chapters, transcription
|
|
from .models import TranscriptionSettings
|
|
|
|
SETTINGS = '/podcasts/transcription/settings/'
|
|
TEST = '/podcasts/transcription/test/'
|
|
|
|
|
|
def fake_response(status=200):
|
|
r = MagicMock()
|
|
r.status_code = status
|
|
r.headers = {}
|
|
r.raw = None # nothing to introspect for the peer check
|
|
return r
|
|
|
|
|
|
class SettingsTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username='alice', password='pw12345678')
|
|
self.client.force_login(self.user)
|
|
|
|
def post(self, data, url=SETTINGS):
|
|
return self.client.post(url, json.dumps(data), content_type='application/json')
|
|
|
|
def test_requires_login(self):
|
|
self.client.logout()
|
|
self.assertEqual(self.client.get(SETTINGS).status_code, 401)
|
|
self.assertEqual(self.post({}, TEST).status_code, 401)
|
|
|
|
def test_defaults(self):
|
|
self.assertEqual(self.client.get(SETTINGS).json(), {
|
|
'engine': 'builtin', 'url': '', 'model': '', 'language': '', 'parallel': 3,
|
|
'has_token': False, 'builtin_available': False})
|
|
self.assertFalse(TranscriptionSettings.objects.exists())
|
|
|
|
@override_settings(WHISPER_URL='http://whisper:8000/v1')
|
|
def test_builtin_available_follows_the_environment_setting(self):
|
|
self.assertTrue(self.client.get(SETTINGS).json()['builtin_available'])
|
|
|
|
def test_roundtrip(self):
|
|
r = self.post({'engine': 'custom', 'url': 'https://whisper.example.org/v1/', 'model': 'large-v3',
|
|
'language': 'de', 'parallel': 5, 'token': 'sk-secret'})
|
|
self.assertEqual(r.status_code, 200)
|
|
want = {'engine': 'custom', 'url': 'https://whisper.example.org/v1', 'model': 'large-v3',
|
|
'language': 'de', 'parallel': 5, 'has_token': True, 'builtin_available': False}
|
|
self.assertEqual(r.json(), want)
|
|
self.assertEqual(self.client.get(SETTINGS).json(), want)
|
|
|
|
def test_token_is_write_only_and_stored_encrypted(self):
|
|
self.post({'engine': 'custom', 'url': 'https://w.example.org/v1', 'token': 'sk-secret'})
|
|
conf = TranscriptionSettings.objects.get(user=self.user)
|
|
self.assertTrue(conf.token_encrypted)
|
|
self.assertNotIn('sk-secret', conf.token_encrypted)
|
|
self.assertEqual(transcription.token_of(conf), 'sk-secret')
|
|
for body in (self.client.get(SETTINGS).content, self.post({'model': 'm'}).content):
|
|
self.assertNotIn(b'sk-secret', body)
|
|
self.assertNotIn(conf.token_encrypted.encode(), body)
|
|
# Not sending a token keeps it; an empty one removes it.
|
|
conf.refresh_from_db()
|
|
self.assertTrue(conf.token_encrypted)
|
|
self.assertFalse(self.post({'token': ''}).json()['has_token'])
|
|
conf.refresh_from_db()
|
|
self.assertEqual(conf.token_encrypted, '')
|
|
|
|
def test_validation(self):
|
|
bad = [
|
|
{'engine': 'cloud'}, {'parallel': 0}, {'parallel': 9}, {'parallel': '3'}, {'parallel': True},
|
|
{'url': 'ftp://x.org/v1'}, {'url': 'javascript:alert(1)'}, {'url': 'whisper.local/v1'},
|
|
{'language': 'Deutsch'}, {'model': 5}, {'token': 5},
|
|
{'engine': 'custom'}, # a custom instance needs a url
|
|
]
|
|
for body in bad:
|
|
self.assertEqual(self.post(body).status_code, 400, body)
|
|
self.assertEqual(self.client.post(SETTINGS, 'nope', content_type='application/json').status_code, 400)
|
|
self.assertFalse(TranscriptionSettings.objects.exists())
|
|
for body in ({'parallel': 1}, {'parallel': 8}, {'language': ''}, {'language': 'pt-BR'}):
|
|
self.assertEqual(self.post(body).status_code, 200, body)
|
|
|
|
def test_settings_are_per_user(self):
|
|
self.post({'parallel': 7})
|
|
other = User.objects.create_user(username='bob', password='pw12345678')
|
|
self.client.force_login(other)
|
|
self.assertEqual(self.client.get(SETTINGS).json()['parallel'], 3)
|
|
|
|
|
|
class TestEndpointTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username='alice', password='pw12345678')
|
|
self.client.force_login(self.user)
|
|
|
|
def configure(self, **kw):
|
|
TranscriptionSettings.objects.update_or_create(user=self.user, defaults=kw)
|
|
|
|
def test_custom_sends_a_second_of_silence(self):
|
|
self.configure(engine='custom', url='https://whisper.example.org/v1', model='large-v3', language='de',
|
|
token_encrypted=logins.encrypt('sk-secret'))
|
|
calls = []
|
|
|
|
def post(url, **kw):
|
|
calls.append((url, kw))
|
|
return fake_response()
|
|
|
|
with patch('podcasts.chapters.assert_safe_url'), patch('podcasts.chapters.requests.post', post):
|
|
r = self.client.post(TEST)
|
|
self.assertEqual(r.json(), {'ok': True, 'model': 'large-v3', 'error': ''})
|
|
url, kw = calls[0]
|
|
self.assertEqual(url, 'https://whisper.example.org/v1/audio/transcriptions')
|
|
self.assertEqual(kw['headers']['Authorization'], 'Bearer sk-secret')
|
|
self.assertEqual(kw['data']['model'], 'large-v3')
|
|
self.assertEqual(kw['data']['language'], 'de')
|
|
name, wav, mime = kw['files']['file']
|
|
self.assertEqual(mime, 'audio/wav')
|
|
with wave.open(io.BytesIO(wav)) as w:
|
|
self.assertEqual((w.getframerate(), w.getnchannels(), w.getnframes()), (16000, 1, 16000))
|
|
|
|
def test_custom_private_address_is_refused(self):
|
|
self.configure(engine='custom', url='http://127.0.0.1:9000/v1')
|
|
with patch('podcasts.chapters.requests.post') as post:
|
|
r = self.client.post(TEST)
|
|
self.assertFalse(r.json()['ok'])
|
|
self.assertTrue(r.json()['error'])
|
|
post.assert_not_called()
|
|
|
|
def test_custom_http_error_is_reported(self):
|
|
self.configure(engine='custom', url='https://whisper.example.org/v1')
|
|
with patch('podcasts.chapters.assert_safe_url'), \
|
|
patch('podcasts.chapters.requests.post', return_value=fake_response(401)):
|
|
r = self.client.post(TEST)
|
|
self.assertEqual((r.json()['ok'], r.json()['error']), (False, 'HTTP 401'))
|
|
|
|
@override_settings(WHISPER_URL='http://whisper:8000/v1')
|
|
def test_builtin_is_exempt_from_the_address_check(self):
|
|
self.configure(engine='builtin')
|
|
with patch('podcasts.transcription.requests.post', return_value=fake_response()) as post, \
|
|
patch('podcasts.chapters.assert_safe_url', side_effect=chapters.WebDAVInputError('private')):
|
|
r = self.client.post(TEST)
|
|
self.assertTrue(r.json()['ok'])
|
|
self.assertEqual(post.call_args.args[0], 'http://whisper:8000/v1/audio/transcriptions')
|
|
self.assertNotIn('Authorization', post.call_args.kwargs['headers'])
|
|
|
|
def test_builtin_without_instance(self):
|
|
r = self.client.post(TEST)
|
|
self.assertFalse(r.json()['ok'])
|
|
self.assertTrue(r.json()['error'])
|