mirror of
https://github.com/aljazceru/plugins.git
synced 2025-12-24 00:24:19 +01:00
ad27d4f624
Now it requires the datastore (or datastore.py plugin), if you want to use runes. The simple reader/writer control lists don't require that, but for simplicity we always require the runes Python module. Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>
286 lines
12 KiB
Python
Executable File
286 lines
12 KiB
Python
Executable File
import os
|
|
from pyln.testing.fixtures import * # noqa: F401,F403
|
|
from pyln.client import RpcError
|
|
import pytest
|
|
import json
|
|
import runes
|
|
import commando
|
|
import time
|
|
|
|
plugin_path = os.path.join(os.path.dirname(__file__), "commando.py")
|
|
datastore_path = os.path.join(os.path.dirname(os.path.dirname(__file__)),
|
|
"datastore", "datastore.py")
|
|
|
|
|
|
def test_add_reader_restrictions():
|
|
mrune = runes.MasterRune(bytes(32))
|
|
runestr = commando.add_reader_restrictions(mrune.copy())
|
|
assert mrune.check_with_reason(runestr, {'method': 'listfoo'}) == (True, '')
|
|
assert mrune.check_with_reason(runestr, {'method': 'getfoo'}) == (True, '')
|
|
assert mrune.check_with_reason(runestr, {'method': 'getsharedsecret'}) == (False, 'method: = getsharedsecret')
|
|
assert mrune.check_with_reason(runestr, {'method': 'summary'}) == (True, '')
|
|
assert mrune.check_with_reason(runestr, {'method': 'fail'}) == (False, 'method: does not start with list AND method: does not start with get AND method: != summary')
|
|
|
|
|
|
def test_commando(node_factory):
|
|
l1, l2 = node_factory.line_graph(2, fundchannel=False)
|
|
|
|
l1.rpc.plugin_start(plugin_path, commando_reader=l2.info['id'])
|
|
l2.rpc.plugin_start(plugin_path)
|
|
|
|
# This works
|
|
res = l2.rpc.call(method='commando',
|
|
payload={'peer_id': l1.info['id'],
|
|
'method': 'listpeers'})
|
|
assert len(res['peers']) == 1
|
|
assert res['peers'][0]['id'] == l2.info['id']
|
|
|
|
res = l2.rpc.call(method='commando',
|
|
payload={'peer_id': l1.info['id'],
|
|
'method': 'listpeers',
|
|
'params': {'id': l2.info['id']}})
|
|
assert len(res['peers']) == 1
|
|
assert res['peers'][0]['id'] == l2.info['id']
|
|
|
|
# This fails
|
|
with pytest.raises(RpcError, match='method: does not start with list AND method: does not start with get AND method: != summary'):
|
|
l2.rpc.call(method='commando',
|
|
payload={'peer_id': l1.info['id'],
|
|
'method': 'withdraw'})
|
|
|
|
# As a writer, anything goes.
|
|
l1.rpc.plugin_stop(plugin_path)
|
|
l1.rpc.plugin_start(plugin_path, commando_writer=l2.info['id'])
|
|
|
|
with pytest.raises(RpcError, match='missing required parameter'):
|
|
l2.rpc.call(method='commando',
|
|
payload={'peer_id': l1.info['id'],
|
|
'method': 'withdraw'})
|
|
|
|
ret = l2.rpc.call(method='commando',
|
|
payload={'peer_id': l1.info['id'],
|
|
'method': 'ping',
|
|
'params': {'id': l2.info['id']}})
|
|
assert 'totlen' in ret
|
|
|
|
# Now, this will go over a single message!
|
|
ret = l2.rpc.call(method='commando',
|
|
payload={'peer_id': l1.info['id'],
|
|
'method': 'getlog',
|
|
'params': {'level': 'io'}})
|
|
|
|
assert len(json.dumps(ret)) > 65535
|
|
|
|
|
|
def test_commando_no_datastore(node_factory):
|
|
l1, l2 = node_factory.line_graph(2, fundchannel=False,
|
|
opts={'plugin': plugin_path})
|
|
# These can happen before other init messages
|
|
l1.daemon.logsearch_start = 0
|
|
l1.daemon.wait_for_log("Initialized without rune support")
|
|
l2.daemon.logsearch_start = 0
|
|
l2.daemon.wait_for_log("Initialized without rune support")
|
|
|
|
with pytest.raises(RpcError, match="No datastore available"):
|
|
l2.rpc.commando_rune(l1.info['id'])
|
|
|
|
|
|
def test_commando_rune(node_factory):
|
|
l1, l2, l3 = node_factory.line_graph(3, fundchannel=False,
|
|
opts={'plugin': [plugin_path,
|
|
datastore_path]})
|
|
|
|
l1.daemon.logsearch_start = 0
|
|
l1.daemon.wait_for_log("Initialized with rune support")
|
|
l2.daemon.logsearch_start = 0
|
|
l2.daemon.wait_for_log("Initialized with rune support")
|
|
l3.daemon.logsearch_start = 0
|
|
l3.daemon.wait_for_log("Initialized with rune support")
|
|
|
|
wrune = l2.rpc.commando_rune()['rune']
|
|
rrune = l2.rpc.commando_rune(restrictions='readonly')['rune']
|
|
|
|
# This works
|
|
res = l1.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'rune': rrune,
|
|
'method': 'listpeers'})
|
|
assert len(res['peers']) == 2
|
|
|
|
# This fails (no rune!)
|
|
with pytest.raises(RpcError, match='Not authorized'):
|
|
l1.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'method': 'withdraw'})
|
|
|
|
# This fails (ro rune!)
|
|
with pytest.raises(RpcError, match='Not authorized'):
|
|
res = l1.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'rune': rrune,
|
|
'method': 'withdraw'})
|
|
|
|
# This would succeed, except missing param)
|
|
with pytest.raises(RpcError, match='missing required parameter'):
|
|
res = l1.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'rune': wrune,
|
|
'method': 'withdraw'})
|
|
|
|
# We can subrune and use that rune explicitly.
|
|
lcrune = l2.rpc.commando_rune(rrune, 'method=listchannels')['rune']
|
|
with pytest.raises(RpcError, match='Not authorized'):
|
|
l1.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'rune': lcrune,
|
|
'method': 'listpeers'})
|
|
|
|
l1.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'rune': lcrune,
|
|
'method': 'listchannels'})
|
|
|
|
# Only allow it to list l3's channels (by source, second param)
|
|
lcrune = l2.rpc.commando_rune(rrune, ['method=listchannels',
|
|
'pnamesource=' + l3.info['id']
|
|
+ '|' + 'parr1=' + l3.info['id']])['rune']
|
|
|
|
# Needs rune!
|
|
with pytest.raises(RpcError, match='Not authorized'):
|
|
l3.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'method': 'listchannels',
|
|
'params': [None, l3.info['id']]})
|
|
# Command wrong
|
|
with pytest.raises(RpcError, match='Not authorized.*method'):
|
|
l3.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'rune': lcrune,
|
|
'method': 'withdraw'})
|
|
|
|
# Params missing
|
|
with pytest.raises(RpcError, match='Not authorized.*missing'):
|
|
l3.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'rune': lcrune,
|
|
'method': 'listchannels'})
|
|
|
|
# Param wrong (array)
|
|
with pytest.raises(RpcError, match='Not authorized.*parr1'):
|
|
l3.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'rune': lcrune,
|
|
'method': 'listchannels',
|
|
'params': [None, l2.info['id']]})
|
|
|
|
# Param wrong (obj)
|
|
with pytest.raises(RpcError, match='Not authorized.*pnamesource'):
|
|
l3.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'rune': lcrune,
|
|
'method': 'listchannels',
|
|
'params': {'source': l2.info['id']}})
|
|
|
|
# Param right (array)
|
|
l3.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'rune': lcrune,
|
|
'method': 'listchannels',
|
|
'params': [None, l3.info['id']]})
|
|
|
|
# Param right (obj)
|
|
l3.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'rune': lcrune,
|
|
'method': 'listchannels',
|
|
'params': {'source': l3.info['id']}})
|
|
|
|
|
|
def test_commando_cacherune(node_factory):
|
|
l1, l2 = node_factory.line_graph(2, fundchannel=False,
|
|
opts={'plugin': [plugin_path,
|
|
datastore_path]})
|
|
restrictions = ['method=listchannels',
|
|
'pnamesource={id}|parr1={id}'.format(id=l1.info['id'])]
|
|
lcrune = l2.rpc.commando_rune(restrictions=restrictions)['rune']
|
|
|
|
# You can't set it, it needs to be via commando!
|
|
with pytest.raises(RpcError,
|
|
match='Must be called as a remote commando call'):
|
|
l1.rpc.commando_cacherune(lcrune)
|
|
|
|
l1.rpc.commando(peer_id=l2.info['id'],
|
|
method='commando-cacherune',
|
|
rune=lcrune)
|
|
|
|
# Param wrong (array)
|
|
with pytest.raises(RpcError, match='Not authorized.*parr1'):
|
|
l1.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'method': 'listchannels',
|
|
'params': [None, l2.info['id']]})
|
|
|
|
# Param wrong (obj)
|
|
with pytest.raises(RpcError, match='Not authorized.*pnamesource'):
|
|
l1.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'method': 'listchannels',
|
|
'params': {'source': l2.info['id']}})
|
|
|
|
# Param right (array)
|
|
l1.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'method': 'listchannels',
|
|
'params': [None, l1.info['id']]})
|
|
|
|
# Param right (obj)
|
|
l1.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'method': 'listchannels',
|
|
'params': {'source': l1.info['id']}})
|
|
|
|
# Still works after restart!
|
|
l2.restart()
|
|
l1.rpc.connect(l2.info['id'], 'localhost', l2.port)
|
|
l1.rpc.call(method='commando',
|
|
payload={'peer_id': l2.info['id'],
|
|
'method': 'listchannels',
|
|
'params': {'source': l1.info['id']}})
|
|
|
|
|
|
def test_rune_time(node_factory):
|
|
l1, l2 = node_factory.line_graph(2, fundchannel=False,
|
|
opts={'plugin': [plugin_path,
|
|
datastore_path]})
|
|
|
|
rune = l1.rpc.commando_rune(restrictions=["method=commando-rune",
|
|
"pnamerestrictions^id=|parr1^id=",
|
|
"time<{}"
|
|
.format(int(time.time()) + 15)])['rune']
|
|
# l2 has to obey restrictions
|
|
with pytest.raises(RpcError, match='Not authorized.*pnamerestrictions'):
|
|
l2.rpc.commando(peer_id=l1.info['id'], method='commando-rune', rune=rune)
|
|
|
|
with pytest.raises(RpcError, match='Not authorized.*pnamerestrictions'):
|
|
l2.rpc.commando(peer_id=l1.info['id'], method='commando-rune', rune=rune,
|
|
params={'restrictions': 'id<{}'.format(l2.info['id'])})
|
|
|
|
# By name
|
|
rune2 = l2.rpc.commando(peer_id=l1.info['id'],
|
|
method='commando-rune',
|
|
rune=rune,
|
|
params={'restrictions': 'id={}'.format(l2.info['id'])})
|
|
# By position
|
|
rune2a = l2.rpc.commando(peer_id=l1.info['id'],
|
|
method='commando-rune',
|
|
rune=rune,
|
|
params=[None, 'id={}'.format(l2.info['id'])])
|
|
assert rune2 == rune2a
|
|
|
|
time.sleep(16)
|
|
with pytest.raises(RpcError, match='Not authorized.*time'):
|
|
l2.rpc.commando(peer_id=l1.info['id'],
|
|
method='commando-rune',
|
|
rune=rune,
|
|
params={'restrictions': 'id={}'.format(l2.info['id'])})
|