mirror of
https://gitlab.com/fdroid/fdroidserver.git
synced 2024-11-04 22:40:12 +01:00
Merge branch 'plugin-system' into 'master'
Plugin system See merge request fdroid/fdroidserver!709
This commit is contained in:
Hans-Christoph Steiner
commit
83edb5b80a
@ -1,6 +1,7 @@
|
||||
#!/usr/bin/env python3
|
||||
#
|
||||
# fdroidserver/__main__.py - part of the FDroid server tools
|
||||
# Copyright (C) 2020 Michael Pöhn <michael.poehn@fsfe.org>
|
||||
# Copyright (C) 2010-2015, Ciaran Gultnieks, ciaran@ciarang.com
|
||||
# Copyright (C) 2013-2014 Daniel Marti <mvdan@mvdan.cc>
|
||||
#
|
||||
@ -17,9 +18,11 @@
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import re
|
||||
import sys
|
||||
import os
|
||||
import locale
|
||||
import pkgutil
|
||||
import logging
|
||||
|
||||
import fdroidserver.common
|
||||
@ -29,7 +32,7 @@ from argparse import ArgumentError
|
||||
from collections import OrderedDict
|
||||
|
||||
|
||||
commands = OrderedDict([
|
||||
COMMANDS = OrderedDict([
|
||||
("build", _("Build a package from source")),
|
||||
("init", _("Quickly start a new repository")),
|
||||
("publish", _("Sign and place packages in the repo")),
|
||||
@ -54,25 +57,85 @@ commands = OrderedDict([
|
||||
])
|
||||
|
||||
|
||||
def print_help():
|
||||
def print_help(available_plugins=None):
|
||||
print(_("usage: ") + _("fdroid [<command>] [-h|--help|--version|<args>]"))
|
||||
print("")
|
||||
print(_("Valid commands are:"))
|
||||
for cmd, summary in commands.items():
|
||||
for cmd, summary in COMMANDS.items():
|
||||
print(" " + cmd + ' ' * (15 - len(cmd)) + summary)
|
||||
if available_plugins:
|
||||
print(_('commands from plugin modules:'))
|
||||
for command in sorted(available_plugins.keys()):
|
||||
print(' {:15}{}'.format(command, available_plugins[command]['summary']))
|
||||
print("")
|
||||
|
||||
|
||||
def preparse_plugin(module_name, module_dir):
|
||||
"""simple regex based parsing for plugin scripts,
|
||||
so we don't have to import them when we just need the summary,
|
||||
but not plan on executing this particular plugin."""
|
||||
if '.' in module_name:
|
||||
raise ValueError("No '.' allowed in fdroid plugin modules: '{}'"
|
||||
.format(module_name))
|
||||
path = os.path.join(module_dir, module_name + '.py')
|
||||
if not os.path.isfile(path):
|
||||
path = os.path.join(module_dir, module_name, '__main__.py')
|
||||
if not os.path.isfile(path):
|
||||
raise ValueError("unable to find main plugin script "
|
||||
"for module '{n}' ('{d}')"
|
||||
.format(n=module_name,
|
||||
d=module_dir))
|
||||
summary = None
|
||||
main = None
|
||||
with open(path, 'r', encoding='utf-8') as f:
|
||||
re_main = re.compile(r'^(\s*def\s+main\s*\(.*\)\s*:'
|
||||
r'|\s*main\s*=\s*lambda\s*:.+)$')
|
||||
re_summary = re.compile(r'^\s*fdroid_summary\s*=\s["\'](?P<text>.+)["\']$')
|
||||
for line in f:
|
||||
m_summary = re_summary.match(line)
|
||||
if m_summary:
|
||||
summary = m_summary.group('text')
|
||||
if re_main.match(line):
|
||||
main = True
|
||||
|
||||
if summary is None:
|
||||
raise NameError("could not find 'fdroid_summary' in: '{}' plugin"
|
||||
.format(module_name))
|
||||
if main is None:
|
||||
raise NameError("could not find 'main' function in: '{}' plugin"
|
||||
.format(module_name))
|
||||
return {'name': module_name, 'summary': summary}
|
||||
|
||||
|
||||
def find_plugins():
|
||||
found_plugins = [{'name': x[1], 'dir': x[0].path} for x in pkgutil.iter_modules() if x[1].startswith('fdroid_')]
|
||||
plugin_infos = {}
|
||||
for plugin_def in found_plugins:
|
||||
command_name = plugin_def['name'][7:]
|
||||
try:
|
||||
plugin_infos[command_name] = preparse_plugin(plugin_def['name'],
|
||||
plugin_def['dir'])
|
||||
except Exception as e:
|
||||
# We need to keep module lookup fault tolerant because buggy
|
||||
# modules must not prevent fdroidserver from functioning
|
||||
if len(sys.argv) > 1 and sys.argv[1] == command_name:
|
||||
# only raise exeption when a user specifies the broken
|
||||
# plugin in explicitly in command line
|
||||
raise e
|
||||
return plugin_infos
|
||||
|
||||
|
||||
def main():
|
||||
available_plugins = find_plugins()
|
||||
|
||||
if len(sys.argv) <= 1:
|
||||
print_help()
|
||||
print_help(available_plugins=available_plugins)
|
||||
sys.exit(0)
|
||||
|
||||
command = sys.argv[1]
|
||||
if command not in commands:
|
||||
if command not in COMMANDS and command not in available_plugins.keys():
|
||||
if command in ('-h', '--help'):
|
||||
print_help()
|
||||
print_help(available_plugins=available_plugins)
|
||||
sys.exit(0)
|
||||
elif command == '--version':
|
||||
output = _('no version info found!')
|
||||
@ -99,11 +162,11 @@ def main():
|
||||
else:
|
||||
from pkg_resources import get_distribution
|
||||
output = get_distribution('fdroidserver').version + '\n'
|
||||
print(output),
|
||||
print(output)
|
||||
sys.exit(0)
|
||||
else:
|
||||
print(_("Command '%s' not recognised.\n" % command))
|
||||
print_help()
|
||||
print_help(available_plugins=available_plugins)
|
||||
sys.exit(1)
|
||||
|
||||
verbose = any(s in sys.argv for s in ['-v', '--verbose'])
|
||||
@ -133,7 +196,10 @@ def main():
|
||||
sys.argv[0] += ' ' + command
|
||||
|
||||
del sys.argv[1]
|
||||
mod = __import__('fdroidserver.' + command, None, None, [command])
|
||||
if command in COMMANDS.keys():
|
||||
mod = __import__('fdroidserver.' + command, None, None, [command])
|
||||
else:
|
||||
mod = __import__(available_plugins[command]['name'], None, None, [command])
|
||||
|
||||
system_langcode, system_encoding = locale.getdefaultlocale()
|
||||
if system_encoding is None or system_encoding.lower() not in ('utf-8', 'utf8'):
|
||||
|
||||
@ -4,8 +4,12 @@ import inspect
|
||||
import optparse
|
||||
import os
|
||||
import sys
|
||||
import pkgutil
|
||||
import textwrap
|
||||
import unittest
|
||||
import tempfile
|
||||
from unittest import mock
|
||||
from testcommon import TmpCwd, TmpPyPath
|
||||
|
||||
localmodule = os.path.realpath(
|
||||
os.path.join(os.path.dirname(inspect.getfile(inspect.currentframe())), '..'))
|
||||
@ -17,12 +21,12 @@ from fdroidserver import common
|
||||
import fdroidserver.__main__
|
||||
|
||||
|
||||
class FdroidTest(unittest.TestCase):
|
||||
class MainTest(unittest.TestCase):
|
||||
'''this tests fdroid.py'''
|
||||
|
||||
def test_commands(self):
|
||||
def test_COMMANDS_check(self):
|
||||
"""make sure the built in sub-command defs didn't change unintentionally"""
|
||||
self.assertListEqual([x for x in fdroidserver.__main__.commands.keys()],
|
||||
self.assertListEqual([x for x in fdroidserver.__main__.COMMANDS.keys()],
|
||||
['build',
|
||||
'init',
|
||||
'publish',
|
||||
@ -49,18 +53,169 @@ class FdroidTest(unittest.TestCase):
|
||||
co = mock.Mock()
|
||||
with mock.patch('sys.argv', ['', 'init', '-h']):
|
||||
with mock.patch('fdroidserver.init.main', co):
|
||||
with mock.patch('sys.exit', lambda x: None):
|
||||
with mock.patch('sys.exit') as exit_mock:
|
||||
fdroidserver.__main__.main()
|
||||
# note: this is sloppy, if `init` changes
|
||||
# this might need changing too
|
||||
exit_mock.assert_called_once_with(0)
|
||||
co.assert_called_once_with()
|
||||
|
||||
def test_call_deploy(self):
|
||||
co = mock.Mock()
|
||||
with mock.patch('sys.argv', ['', 'deploy', '-h']):
|
||||
with mock.patch('fdroidserver.server.main', co):
|
||||
with mock.patch('sys.exit', lambda x: None):
|
||||
with mock.patch('sys.exit') as exit_mock:
|
||||
fdroidserver.__main__.main()
|
||||
# note: this is sloppy, if `deploy` changes
|
||||
# this might need changing too
|
||||
exit_mock.assert_called_once_with(0)
|
||||
co.assert_called_once_with()
|
||||
|
||||
def test_find_plugins(self):
|
||||
with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir):
|
||||
with open('fdroid_testy1.py', 'w') as f:
|
||||
f.write(textwrap.dedent("""\
|
||||
fdroid_summary = "ttt"
|
||||
main = lambda: 'all good'"""))
|
||||
with TmpPyPath(tmpdir):
|
||||
plugins = fdroidserver.__main__.find_plugins()
|
||||
self.assertIn('testy1', plugins.keys())
|
||||
self.assertEqual(plugins['testy1']['summary'], 'ttt')
|
||||
self.assertEqual(__import__(plugins['testy1']['name'],
|
||||
None,
|
||||
None,
|
||||
['testy1'])
|
||||
.main(),
|
||||
'all good')
|
||||
|
||||
def test_main_plugin_lambda(self):
|
||||
with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir):
|
||||
with open('fdroid_testy2.py', 'w') as f:
|
||||
f.write(textwrap.dedent("""\
|
||||
fdroid_summary = "ttt"
|
||||
main = lambda: print('all good')"""))
|
||||
with TmpPyPath(tmpdir):
|
||||
with mock.patch('sys.argv', ['', 'testy2']):
|
||||
with mock.patch('sys.exit') as exit_mock:
|
||||
fdroidserver.__main__.main()
|
||||
exit_mock.assert_called_once_with(0)
|
||||
|
||||
def test_main_plugin_def(self):
|
||||
with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir):
|
||||
with open('fdroid_testy3.py', 'w') as f:
|
||||
f.write(textwrap.dedent("""\
|
||||
fdroid_summary = "ttt"
|
||||
def main():
|
||||
print('all good')"""))
|
||||
with TmpPyPath(tmpdir):
|
||||
with mock.patch('sys.argv', ['', 'testy3']):
|
||||
with mock.patch('sys.exit') as exit_mock:
|
||||
fdroidserver.__main__.main()
|
||||
exit_mock.assert_called_once_with(0)
|
||||
|
||||
def test_main_broken_plugin(self):
|
||||
"""making sure broken plugins get their exceptions through"""
|
||||
with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir):
|
||||
with open('fdroid_testy4.py', 'w') as f:
|
||||
f.write(textwrap.dedent("""\
|
||||
fdroid_summary = "ttt"
|
||||
def main():
|
||||
raise Exception("this plugin is broken")"""))
|
||||
with TmpPyPath(tmpdir):
|
||||
with mock.patch('sys.argv', ['', 'testy4']):
|
||||
with self.assertRaisesRegex(Exception, "this plugin is broken"):
|
||||
fdroidserver.__main__.main()
|
||||
|
||||
def test_main_malicious_plugin(self):
|
||||
"""The purpose of this test is to make sure code in plugins
|
||||
doesn't get executed unintentionally.
|
||||
"""
|
||||
with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir):
|
||||
with open('fdroid_testy5.py', 'w') as f:
|
||||
f.write(textwrap.dedent("""\
|
||||
fdroid_summary = "ttt"
|
||||
raise Exception("this plugin is malicious")
|
||||
def main():
|
||||
print("evil things")"""))
|
||||
with TmpPyPath(tmpdir):
|
||||
with mock.patch('sys.argv', ['', 'lint']):
|
||||
with mock.patch('sys.exit') as exit_mock:
|
||||
fdroidserver.__main__.main()
|
||||
# note: this is sloppy, if `lint` changes
|
||||
# this might need changing too
|
||||
exit_mock.assert_called_once_with(0)
|
||||
|
||||
def test_main_prevent_plugin_override(self):
|
||||
"""making sure build-in subcommands cannot be overridden by plugins
|
||||
"""
|
||||
with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir):
|
||||
with open('fdroid_signatures.py', 'w') as f:
|
||||
f.write(textwrap.dedent("""\
|
||||
fdroid_summary = "ttt"
|
||||
def main():
|
||||
raise("plugin overrides don't get prevent!")"""))
|
||||
with TmpPyPath(tmpdir):
|
||||
with mock.patch('sys.argv', ['', 'signatures']):
|
||||
with mock.patch('sys.exit') as exit_mock:
|
||||
fdroidserver.__main__.main()
|
||||
# note: this is sloppy, if `signatures` changes
|
||||
# this might need changing too
|
||||
self.assertEqual(exit_mock.call_count, 2)
|
||||
|
||||
def test_preparse_plugin_lookup_bad_name(self):
|
||||
self.assertRaises(ValueError,
|
||||
fdroidserver.__main__.preparse_plugin,
|
||||
"some.package", "/non/existent/module/path")
|
||||
|
||||
def test_preparse_plugin_lookup_bad_path(self):
|
||||
self.assertRaises(ValueError,
|
||||
fdroidserver.__main__.preparse_plugin,
|
||||
"fake_module_name", "/non/existent/module/path")
|
||||
|
||||
def test_preparse_plugin_lookup_summary_missing(self):
|
||||
with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir):
|
||||
with open('fdroid_testy6.py', 'w') as f:
|
||||
f.write(textwrap.dedent("""\
|
||||
main = lambda: print('all good')"""))
|
||||
with TmpPyPath(tmpdir):
|
||||
p = [x for x in pkgutil.iter_modules() if x[1].startswith('fdroid_')]
|
||||
module_dir = p[0][0].path
|
||||
module_name = p[0][1]
|
||||
self.assertRaises(NameError,
|
||||
fdroidserver.__main__.preparse_plugin,
|
||||
module_name, module_dir)
|
||||
|
||||
def test_preparse_plugin_lookup_module_file(self):
|
||||
with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir):
|
||||
with open('fdroid_testy7.py', 'w') as f:
|
||||
f.write(textwrap.dedent("""\
|
||||
fdroid_summary = "ttt"
|
||||
main = lambda: pritn('all good')"""))
|
||||
with TmpPyPath(tmpdir):
|
||||
p = [x for x in pkgutil.iter_modules() if x[1].startswith('fdroid_')]
|
||||
module_path = p[0][0].path
|
||||
module_name = p[0][1]
|
||||
d = fdroidserver.__main__.preparse_plugin(module_name, module_path)
|
||||
self.assertDictEqual(d, {'name': 'fdroid_testy7',
|
||||
'summary': 'ttt'})
|
||||
|
||||
def test_preparse_plugin_lookup_module_dir(self):
|
||||
with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir):
|
||||
os.mkdir(os.path.join(tmpdir, 'fdroid_testy8'))
|
||||
with open('fdroid_testy8/__main__.py', 'w') as f:
|
||||
f.write(textwrap.dedent("""\
|
||||
fdroid_summary = "ttt"
|
||||
main = lambda: print('all good')"""))
|
||||
with open('fdroid_testy8/__init__.py', 'w') as f:
|
||||
pass
|
||||
with TmpPyPath(tmpdir):
|
||||
p = [x for x in pkgutil.iter_modules() if x[1].startswith('fdroid_')]
|
||||
module_path = p[0][0].path
|
||||
module_name = p[0][1]
|
||||
d = fdroidserver.__main__.preparse_plugin(module_name, module_path)
|
||||
self.assertDictEqual(d, {'name': 'fdroid_testy8',
|
||||
'summary': 'ttt'})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
os.chdir(os.path.dirname(__file__))
|
||||
@ -71,5 +226,5 @@ if __name__ == "__main__":
|
||||
(common.options, args) = parser.parse_args(['--verbose'])
|
||||
|
||||
newSuite = unittest.TestSuite()
|
||||
newSuite.addTest(unittest.makeSuite(FdroidTest))
|
||||
newSuite.addTest(unittest.makeSuite(MainTest))
|
||||
unittest.main(failfast=False)
|
||||
|
||||
@ -16,6 +16,7 @@
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
|
||||
class TmpCwd():
|
||||
@ -32,3 +33,18 @@ class TmpCwd():
|
||||
|
||||
def __exit__(self, a, b, c):
|
||||
os.chdir(self.orig_cwd)
|
||||
|
||||
|
||||
class TmpPyPath():
|
||||
"""Context-manager for temporarily changing the current working
|
||||
directory.
|
||||
"""
|
||||
|
||||
def __init__(self, additional_path):
|
||||
self.additional_path = additional_path
|
||||
|
||||
def __enter__(self):
|
||||
sys.path.append(self.additional_path)
|
||||
|
||||
def __exit__(self, a, b, c):
|
||||
sys.path.remove(self.additional_path)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user