239 lines
11 KiB
Python
239 lines
11 KiB
Python
import json
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime, timedelta
|
|
from urllib.error import HTTPError
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
|
|
from conftest import PASSWORD, post
|
|
from test_workflow import action, approve, form, new, order, pending, save
|
|
|
|
HOOK = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=TEST-ONLY-NOT-A-REAL-KEY'
|
|
BASE = 'http://192.168.1.10:8080'
|
|
|
|
|
|
def configure(clients, **overrides):
|
|
data = dict(enabled='1', base_url=BASE,
|
|
approval_webhook=HOOK + '-approval', result_webhook=HOOK + '-result')
|
|
data.update(overrides)
|
|
return post(clients['admin'], '/settings/notifications', data)
|
|
|
|
|
|
def notices(app):
|
|
from models import Notification
|
|
with app.Session() as db:
|
|
return [dict(id=n.id, status=n.status, channel=n.channel, attempts=n.attempts, error=n.last_error)
|
|
for n in db.query(Notification).order_by(Notification.id)]
|
|
|
|
|
|
def test_workflow_queues_only_committed_transitions(app, clients, monkeypatch):
|
|
import notifications
|
|
from models import Notification, NotificationSettings
|
|
monkeypatch.setattr(notifications, 'send_wecom', lambda *args: pytest.fail('HTTP request must not send'))
|
|
oid = pending(app, clients) # Disabled by default; no historical replay.
|
|
assert not notices(app)
|
|
assert configure(clients).status_code == 302
|
|
assert action(app, clients['maker'], oid, 'withdraw', comment='fix').status_code == 302
|
|
assert action(app, clients['maker'], oid, 'submit').status_code == 302
|
|
stale = order(app, oid)['version']
|
|
assert action(app, clients['approver'], oid, 'approve', stamped='1').status_code == 302
|
|
assert post(clients['reviewer'], f'/orders/{oid}/approve', {'row_version': str(stale), 'stamped': '1'}).status_code == 409
|
|
assert [n['channel'] for n in notices(app)] == ['result', 'approval', 'result']
|
|
fresh = new(app, clients['maker'])
|
|
assert save(app, clients['maker'], fresh, intent='submit', qty='-1').status_code == 400
|
|
assert len(notices(app)) == 3
|
|
# Even a failure after enqueue rolls back both order and notification.
|
|
original = notifications.enqueue_transition
|
|
def fail_after_enqueue(*args):
|
|
original(*args)
|
|
raise ValueError('simulated persistence failure')
|
|
monkeypatch.setattr(notifications, 'enqueue_transition', fail_after_enqueue)
|
|
assert save(app, clients['maker'], fresh, intent='submit').status_code == 400
|
|
assert order(app, fresh)['status'] == 'draft'
|
|
assert len(notices(app)) == 3
|
|
|
|
|
|
def test_worker_cancels_old_tasks_and_sends_minimal_lan_links(app, clients):
|
|
from notifications import dispatch_once
|
|
configure(clients, result_webhook='')
|
|
oid = pending(app, clients)
|
|
assert action(app, clients['maker'], oid, 'withdraw', comment='修改').status_code == 302
|
|
assert save(app, clients['maker'], oid, intent='submit').status_code == 302
|
|
sent = []
|
|
def sender(hook, payload):
|
|
# Sending must not hold SQLite's write lock.
|
|
with app.Session() as db:
|
|
db.execute(text('BEGIN IMMEDIATE'))
|
|
sent.append((hook, payload))
|
|
now = datetime.utcnow() + timedelta(seconds=1)
|
|
assert dispatch_once(app.Session, sender, now)
|
|
assert notices(app)[0]['status'] == 'cancelled'
|
|
assert not sent
|
|
assert dispatch_once(app.Session, sender, now)
|
|
assert notices(app)[1]['status'] == 'sent'
|
|
assert sent[0][0] == HOOK + '-approval'
|
|
content = sent[0][1]['markdown']['content']
|
|
assert f'{BASE}/orders/{oid}/view' in content
|
|
assert 'Audit supplier' not in content and 'key=' not in content
|
|
assert sent[0][1]['msgtype'] == 'markdown'
|
|
configure(clients)
|
|
assert action(app, clients['approver'], oid, 'approve', stamped='1').status_code == 302
|
|
assert not dispatch_once(app.Session, sender, now) # Shared rate limit.
|
|
assert dispatch_once(app.Session, sender, now + timedelta(seconds=5))
|
|
assert sent[-1][0] == HOOK + '-result'
|
|
assert clients['maker'].get(f'/orders/{oid}/pdf').data.startswith(b'%PDF')
|
|
|
|
|
|
def test_retry_recovery_pause_and_no_credentials_in_errors(app, clients):
|
|
from models import Notification, NotificationSettings
|
|
from notifications import dispatch_once
|
|
configure(clients)
|
|
oid = pending(app, clients)
|
|
now = datetime.utcnow() + timedelta(seconds=1)
|
|
def fail(*args):
|
|
raise RuntimeError(HOOK)
|
|
for attempt in range(5):
|
|
assert dispatch_once(app.Session, fail, now + timedelta(hours=attempt))
|
|
row = notices(app)[0]
|
|
assert row['status'] == 'failed' and row['attempts'] == 5
|
|
assert HOOK not in row['error']
|
|
assert post(clients['admin'], f"/settings/notifications/{row['id']}/retry").status_code == 302
|
|
configure(clients, enabled='0')
|
|
assert not dispatch_once(app.Session, fail, now + timedelta(hours=6))
|
|
configure(clients)
|
|
# Recover a crashed sender after its lease expires.
|
|
with app.Session() as db:
|
|
notice = db.get(Notification, row['id'])
|
|
notice.status = 'sending'
|
|
notice.lease_token = 'expired'
|
|
notice.lease_until = now
|
|
db.commit()
|
|
sent = []
|
|
assert dispatch_once(app.Session, lambda *a: sent.append(a), now + timedelta(hours=6))
|
|
assert len(sent) == 1 and notices(app)[0]['status'] == 'sent'
|
|
assert not dispatch_once(app.Session, fail, now + timedelta(hours=7))
|
|
assert post(clients['admin'], f"/settings/notifications/{row['id']}/retry").status_code == 409
|
|
html = clients['admin'].get('/settings/notifications').get_data(as_text=True)
|
|
assert HOOK not in html and '已发送' in html
|
|
|
|
|
|
def test_parallel_workers_claim_once(app, clients):
|
|
from notifications import dispatch_once
|
|
configure(clients)
|
|
pending(app, clients)
|
|
sent = []
|
|
now = datetime.utcnow() + timedelta(seconds=1)
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
results = list(pool.map(lambda _: dispatch_once(app.Session, lambda *a: sent.append(a), now), range(4)))
|
|
assert results.count(True) == 1 and len(sent) == 1
|
|
|
|
|
|
def test_settings_permissions_secret_preservation_and_testing(app, clients):
|
|
from models import NotificationSettings
|
|
for name in ('maker', 'reviewer', 'approver'):
|
|
assert clients[name].get('/settings/notifications').status_code == 403
|
|
for path in ('/settings/notifications', '/settings/notifications/test', '/settings/notifications/1/retry'):
|
|
assert post(clients[name], path).status_code == 403
|
|
assert clients['admin'].post('/settings/notifications', data={'enabled': '1'}).status_code == 400
|
|
assert configure(clients).status_code == 302
|
|
assert configure(clients, approval_webhook='', result_webhook='').status_code == 302
|
|
with app.Session() as db:
|
|
assert db.get(NotificationSettings, 1).approval_webhook == HOOK + '-approval'
|
|
html = clients['admin'].get('/settings/notifications')
|
|
assert html.headers['Cache-Control'] == 'no-store'
|
|
assert HOOK not in html.get_data(as_text=True)
|
|
for _ in range(2):
|
|
assert post(clients['admin'], '/settings/notifications/test', {'channel': 'approval'}).status_code == 302
|
|
assert len(notices(app)) == 1
|
|
assert configure(clients, clear_approval='1').status_code == 400
|
|
with app.Session() as db:
|
|
assert db.get(NotificationSettings, 1).approval_webhook == HOOK + '-approval'
|
|
assert configure(clients, clear_result='1').status_code == 302
|
|
assert post(clients['admin'], '/settings/notifications/test', {'channel': 'result'}).status_code == 400
|
|
|
|
|
|
@pytest.mark.parametrize('url', ['http://127.0.0.1:8080', 'http://localhost:8080', 'http://0.0.0.0',
|
|
'https://example.org/orders/1/view', 'https://user:pass@192.168.1.2', 'http://192.168.1.2/?x=1',
|
|
'http://192.168.1.2/#x', 'javascript:alert(1)', 'http://8.8.8.8', 'http://192.168.1.2:99999'])
|
|
def test_invalid_lan_addresses(app, clients, url):
|
|
assert configure(clients, base_url=url).status_code == 400
|
|
|
|
|
|
@pytest.mark.parametrize('url', ['http://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=x',
|
|
'https://qyapi.weixin.qq.com.evil.invalid/cgi-bin/webhook/send?key=x',
|
|
HOOK + '&redirect=evil', HOOK + '#secret', 'http://192.168.1.1'])
|
|
def test_webhook_rejects_untrusted_destinations(app, clients, url):
|
|
assert configure(clients, approval_webhook=url).status_code == 400
|
|
assert 'value="' + url + '"' not in clients['admin'].get('/settings/notifications').get_data(as_text=True)
|
|
|
|
|
|
@pytest.mark.parametrize('reply,expected', [(b'{"errcode":0,"errmsg":"ok"}', None),
|
|
(b'{"errcode":45009,"errmsg":"secret url"}', '45009'), (b'{}', '格式'),
|
|
(b'{"errcode":false}', '格式'), (b'not json', '格式'), (b'x'*16385, '异常')])
|
|
def test_wecom_protocol(app, monkeypatch, reply, expected):
|
|
import notifications
|
|
class Response:
|
|
status = 200
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): pass
|
|
def read(self, count): return reply[:count]
|
|
class Opener:
|
|
def open(self, req, timeout):
|
|
assert req.full_url == HOOK and req.method == 'POST' and timeout == 5
|
|
assert json.loads(req.data)['msgtype'] == 'markdown'
|
|
return Response()
|
|
monkeypatch.setattr(notifications, 'build_opener', lambda *args: Opener())
|
|
if expected:
|
|
with pytest.raises(notifications.DeliveryError, match=expected) as error:
|
|
notifications.send_wecom(HOOK, {'msgtype': 'markdown'})
|
|
assert 'secret url' not in str(error.value)
|
|
else:
|
|
notifications.send_wecom(HOOK, {'msgtype': 'markdown'})
|
|
|
|
|
|
def test_wecom_redirect_is_not_followed(app, monkeypatch):
|
|
import notifications
|
|
class Opener:
|
|
def open(self, req, timeout):
|
|
raise HTTPError(HOOK, 302, 'redirect', {}, None)
|
|
def opener(handler):
|
|
assert handler.redirect_request(None, None, 302, '', {}, 'https://evil.invalid') is None
|
|
return Opener()
|
|
monkeypatch.setattr(notifications, 'build_opener', opener)
|
|
with pytest.raises(notifications.DeliveryError, match='302'):
|
|
notifications.send_wecom(HOOK, {})
|
|
|
|
|
|
def test_notice_login_password_return_and_scope(app, clients):
|
|
oid = pending(app, clients)
|
|
destination = f'/orders/{oid}/view'
|
|
with app.Session() as db:
|
|
db.query(app.User).filter_by(username='reviewer').one().must_change_password = True
|
|
db.commit()
|
|
c = app.app.test_client()
|
|
response = c.get(destination)
|
|
assert parse_qs(urlsplit(response.location).query)['next'] == [destination]
|
|
c.get(response.location)
|
|
response = post(c, '/login', {'username': 'reviewer', 'password': PASSWORD, 'next': destination})
|
|
assert urlsplit(response.location).path == '/account/password'
|
|
assert parse_qs(urlsplit(response.location).query)['next'] == [destination]
|
|
c.get(response.location)
|
|
new_password = PASSWORD + '-personal'
|
|
response = post(c, '/account/password', {'old_password': PASSWORD, 'password': new_password,
|
|
'confirm_password': new_password, 'next': destination})
|
|
assert response.location == destination
|
|
assert c.get(destination).status_code == 200
|
|
assert order(app, oid)['status'] == 'pending_approval' # Viewing cannot approve.
|
|
assert clients['other'].get(destination).status_code == 404
|
|
assert c.get(f'/orders/{oid}/review').status_code == 405
|
|
|
|
|
|
@pytest.mark.parametrize('destination', ['https://evil.invalid', '//evil.invalid', '/\\evil.invalid',
|
|
'/orders/1/approve', '/settings/notifications'])
|
|
def test_login_next_cannot_escape_or_execute(app, destination):
|
|
c = app.app.test_client()
|
|
c.get('/login')
|
|
assert post(c, '/login', {'username': 'maker', 'password': PASSWORD, 'next': destination}).location == '/'
|