diff --git a/.github/workflows/cross-platform-ci.yml b/.github/workflows/cross-platform-ci.yml index c233615..b1cc8f7 100644 --- a/.github/workflows/cross-platform-ci.yml +++ b/.github/workflows/cross-platform-ci.yml @@ -6,14 +6,27 @@ on: pull_request: workflow_dispatch: +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +defaults: + run: + shell: bash + jobs: test: runs-on: ${{ matrix.os }} strategy: fail-fast: false matrix: - os: [ubuntu-24.04, ubuntu-22.04, macos-latest, macos-15-intel, windows-latest] - python-version: ['3.10', '3.11', '3.12'] + # oldest and newest Python on every OS, the ones in between on Linux only + os: [ubuntu-24.04, macos-latest, macos-15-intel, windows-latest] + python-version: ['3.10', '3.14'] + include: + - {os: ubuntu-24.04, python-version: '3.11'} + - {os: ubuntu-24.04, python-version: '3.12'} + - {os: ubuntu-24.04, python-version: '3.13'} steps: - uses: actions/checkout@v4 @@ -26,7 +39,7 @@ jobs: - name: Install package run: | python -m pip install --upgrade pip - pip install . + pip install . pytest - name: Verify installed version matches source run: | @@ -41,7 +54,6 @@ jobs: print(f'Installed: {installed}') assert expected == installed, f'Version mismatch: {expected} != {installed}' " - shell: bash - name: Test CLI help run: modcli --help @@ -54,3 +66,66 @@ jobs: - name: Test config commands run: modcli config --help + + # Outside the checkout, so the tests import the installed package (and catch files it doesn't ship) + - name: Run tests + run: | + cp -r tests pyproject.toml "$RUNNER_TEMP/" + cd "$RUNNER_TEMP" && pytest + + uv-install: + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-24.04, macos-latest, macos-15-intel, windows-latest] + + steps: + - uses: actions/checkout@v4 + + - name: Set up uv + uses: astral-sh/setup-uv@v6 + + - name: Install with uv + run: uv tool install . + + - name: Test CLI help + run: modcli --help + + - name: Test auth commands + run: modcli auth --help + + - name: Test bundle commands + run: modcli bundle --help + + - name: Test config commands + run: modcli config --help + + # Outside the checkout, so the tests import the package as uv builds and installs it + - name: Run tests + run: | + cp -r tests pyproject.toml "$RUNNER_TEMP/" + cd "$RUNNER_TEMP" && uvx --with "$GITHUB_WORKSPACE" pytest + + # The oldest versions pyproject.toml allows (click>=8.0) on the oldest Python + lowest-deps: + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v4 + + - name: Set up uv + uses: astral-sh/setup-uv@v6 + + - name: Run tests + run: uv run --resolution lowest-direct --python 3.10 pytest + + lint: + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v4 + + - name: Set up uv + uses: astral-sh/setup-uv@v6 + + - name: Run pylint + run: uv run pylint --errors-only modcli tests diff --git a/README.md b/README.md index 1d19e02..6357547 100644 --- a/README.md +++ b/README.md @@ -25,3 +25,22 @@ Commands: bundle LV2 bundle commands config Configuration commands ``` + +## Development + +``` +uv sync # install with the dev group (pylint, pytest) +uv run pytest # run the tests +uv run pylint --errors-only modcli tests # lint +``` + +## Authentication + +``` +modcli auth login # username and password +modcli auth login-forum # MOD Forum account (login-sso still works) +modcli auth login-google # Google account +modcli auth login-magiclink # sign-in link sent by email +``` + +`login-forum` and `login-google` open the browser, and `login-magiclink` emails a sign-in link. Once you sign in, the browser hands the token back to the CLI automatically. When the browser runs on another machine (e.g. the CLI runs over SSH), paste the token the browser shows into the terminal instead. diff --git a/modcli/auth.py b/modcli/auth.py index ead6404..d1f6d00 100644 --- a/modcli/auth.py +++ b/modcli/auth.py @@ -1,12 +1,6 @@ -import socket -import sys -from http.server import BaseHTTPRequestHandler, HTTPServer from importlib.metadata import version from urllib import parse -import click -from click import Abort - from modcli import http @@ -21,74 +15,24 @@ def login(username: str, password: str, api_url: str): return result.json()['message'].strip() -def get_open_port(): - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.bind(('', 0)) - s.listen(1) - port = s.getsockname()[1] - s.close() - return port - - -def login_sso_detached(api_url: str): - click.echo('Running in detached mode...') - click.echo(f'1) Open this url in any browser: {api_url}/users/tokens_sso') - click.echo('2) The URL will automatically redirect to MOD Forum (https://forum.moddevices.com)') - click.echo('3) Once MOD Forum page loads, if asked, enter your credentials or register a new user') - click.echo('4) A JWT token will be displayed in your browser') - try: - token = click.prompt('Copy the token value and paste it here, then press ENTER') - return token.strip() - except Abort: - sys.exit(1) - - -def login_sso(api_url: str): - server_host = 'localhost' - server_port = get_open_port() - local_server = f'http://{server_host}:{server_port}' - - class SSORequestHandler(BaseHTTPRequestHandler): - token = '' - - def do_HEAD(self): - self.send_response(200) - self.send_header('Content-type', 'text/html') - self.end_headers() +def forum_url(api_url: str, local_url: str, state: str): + query = parse.urlencode({'local_url': local_url, 'state': state}) + return f'{api_url}/users/tokens_modforum?{query}' - def do_GET(self): - response = self.handle_http(200) - _, _, _, query, _ = parse.urlsplit(self.path) - result = parse.parse_qs(query) - tokens = result.get('token', None) - SSORequestHandler.token = tokens[0] if len(tokens) > 0 else None - self.wfile.write(response) - def handle_http(self, status_code): - self.send_response(status_code) - self.send_header('Content-type', 'text/html') - self.end_headers() - content = ''' - modcli - success - Authentication successful! This browser window can be closed. - ''' - return bytes(content, 'UTF-8') +def google_url(api_url: str, local_url: str, state: str): + query = parse.urlencode({'aud': 'api', 'local_url': local_url, 'state': state}) + return f'{api_url}/users/tokens_google?{query}' - def log_message(self, fmt, *args): - pass - httpd = HTTPServer((server_host, server_port), SSORequestHandler) - httpd.timeout = 30 - - click.echo('Open this URL in your browser to authenticate:') - click.echo(f'{api_url}/users/tokens_sso?local_url={local_server}') - - try: - httpd.handle_request() - except KeyboardInterrupt: - pass - - token = SSORequestHandler.token - if not token: - raise Exception('Authentication failed!') - return token +def send_magic_link(api_url: str, email: str, local_url: str, state: str) -> int: + '''Returns how long the link lives, in seconds.''' + result = http.post(f'{api_url}/users/tokens_magiclink', json_data={ + 'email': email, + 'aud': 'api', + 'local_url': local_url, + 'state': state, + }) + if result.status_code != 200: + raise Exception(f'Error: {result.json()["error-message"]}') + return result.json()['expires_in'] diff --git a/modcli/cli.py b/modcli/cli.py index d55dd84..af359cd 100644 --- a/modcli/cli.py +++ b/modcli/cli.py @@ -1,12 +1,18 @@ +import secrets import sys +import webbrowser import click from importlib.metadata import version -from modcli import context, auth, bundle +from modcli import context, auth, bundle, token_receiver -_sso_disclaimer = '''SSO login requires you have a valid account in MOD Forum (https://forum.moddevices.com). -If your browser has an active session the credentials will be used for this login. Confirm?''' +# seconds to wait for the token from the browser; a magic link's wait is the link's validity, from the API +_FORUM_TIMEOUT = 5 * 60 +_GOOGLE_TIMEOUT = 5 * 60 + +# -y no longer does anything; kept so existing scripts passing it keep working +_ignored_confirm_all = click.option('-y', '--confirm-all', is_flag=True, hidden=True, expose_value=False) def _normalize_command_name(name): @@ -39,28 +45,57 @@ def config_group(): pass -@click.command(help='Authenticate user with SSO (MOD Forum)') -@click.option('-s', '--show-token', type=bool, help='Print the JWT token obtained', is_flag=True) -@click.option('-o', '--one-time', type=bool, help='Only print token once (do not store it)', is_flag=True) -@click.option('-y', '--confirm-all', type=bool, help='Confirm all operations', is_flag=True) -@click.option('-d', '--detached-mode', type=bool, help='Run process without opening a local browser', is_flag=True) -@click.option('-e', '--env-name', type=str, help='Switch to environment before authenticating') -def login_sso(show_token: bool, one_time: bool, confirm_all: bool, detached_mode: bool, env_name: str): +def _open_browser(url: str): + click.echo('Sign in in the browser. If it doesn\'t open, visit:') + click.echo(url) + webbrowser.open(url) + + +def _start_forum(api_url: str, local_url: str): + state = secrets.token_urlsafe(16) + click.echo('Sign in with your MOD Forum account (https://forum.mod.audio); an existing browser session is used.') + _open_browser(auth.forum_url(api_url, local_url, state)) + return state, _FORUM_TIMEOUT + + +def _start_google(api_url: str, local_url: str): + state = secrets.token_urlsafe(16) + _open_browser(auth.google_url(api_url, local_url, state)) + return state, _GOOGLE_TIMEOUT + + +def _start_magiclink(email: str): + def start(api_url: str, local_url: str): + state = secrets.token_urlsafe(16) + expires_in = auth.send_magic_link(api_url, email, local_url, state) + click.echo(f'A sign-in link was sent to {email}. Open it and confirm.') + click.echo('No email? Only one link is sent per minute, so wait a minute before trying again.') + return state, expires_in + return start + + +def _receive_token(api_url: str, start): + # start() kicks off the sign-in and returns the state the redirect must carry back, and how long to wait + port = token_receiver.get_open_port() + state, timeout = start(api_url, token_receiver.local_url(port)) + click.echo('Finish in the browser, or paste the token it shows here and press ENTER: ', nl=False) + token, source = token_receiver.receive_token(port, timeout, state) + if source != 'paste': + click.echo() + if not token: + raise Exception('Authentication failed!') + return token + + +def _browser_login(env_name: str, show_token: bool, one_time: bool, start): if env_name: context.set_active_env(env_name) env = context.current_env() - if not confirm_all: - response = click.confirm(_sso_disclaimer) - if not response: - sys.exit(1) if not one_time: click.echo(f'Logging in to [{env.name}]...') try: - if detached_mode: - token = auth.login_sso_detached(env.api_url) - else: - token = auth.login_sso(env.api_url) + token = _receive_token(env.api_url, start) except Exception as ex: click.echo(click.style(str(ex), fg='red'), err=True) sys.exit(1) @@ -76,12 +111,51 @@ def login_sso(show_token: bool, one_time: bool, confirm_all: bool, detached_mode click.echo(click.style(f'You\'re now logged in as [{env.username}] in [{env.name}].', fg='green')) +def _browser_login_options(command): + options = [ + click.option('-s', '--show-token', type=bool, help='Print the JWT token obtained', is_flag=True), + click.option('-o', '--one-time', type=bool, help='Only print token once (do not store it)', is_flag=True), + click.option('-e', '--env-name', type=str, help='Switch to environment before authenticating'), + _ignored_confirm_all, + ] + for option in reversed(options): + command = option(command) + return command + + +@click.command(help='Authenticate user with a MOD Forum account') +@_browser_login_options +def login_forum(show_token: bool, one_time: bool, env_name: str): + _browser_login(env_name, show_token, one_time, _start_forum) + + +# The command's former name, kept for scripts +@click.command(name='login-sso', hidden=True, help='Same as login-forum') +@_browser_login_options +def login_sso(show_token: bool, one_time: bool, env_name: str): + _browser_login(env_name, show_token, one_time, _start_forum) + + +@click.command(help='Authenticate user with Google') +@_browser_login_options +def login_google(show_token: bool, one_time: bool, env_name: str): + _browser_login(env_name, show_token, one_time, _start_google) + + +@click.command(help='Authenticate user with a sign-in link sent by email') +@click.option('--email', type=str, prompt=True, help='Email address of the user') +@_browser_login_options +def login_magiclink(email: str, show_token: bool, one_time: bool, env_name: str): + _browser_login(env_name, show_token, one_time, _start_magiclink(email)) + + @click.command(help='Authenticate user') @click.option('-u', '--username', type=str, prompt=True, help='User ID') @click.option('-p', '--password', type=str, prompt=True, hide_input=True, help='User password') @click.option('-s', '--show-token', type=bool, help='Print the JWT token obtained', is_flag=True) @click.option('-o', '--one-time', type=bool, help='Only print token once (do not store it)', is_flag=True) @click.option('-e', '--env-name', type=str, help='Switch to environment before authenticating') +@_ignored_confirm_all def login(username: str, password: str, show_token: bool, one_time: bool, env_name: str): if env_name: context.set_active_env(env_name) @@ -192,7 +266,10 @@ def publish(project_file: str, packages_path: str, show_result: bool, keep_envir auth_group.add_command(active_token) auth_group.add_command(login) +auth_group.add_command(login_forum) auth_group.add_command(login_sso) +auth_group.add_command(login_google) +auth_group.add_command(login_magiclink) bundle_group.add_command(publish) config_group.add_command(add_env) config_group.add_command(set_active_env) diff --git a/modcli/config.py b/modcli/config.py index 6a31bf5..e7c2119 100644 --- a/modcli/config.py +++ b/modcli/config.py @@ -149,8 +149,8 @@ def __init__(self, name: str, api_url: str, bundle_url: str): def set_token(self, token: str): _, payload, _ = token.split('.') - payload_data = json.loads(base64.b64decode(payload + '===').decode()) - username = payload_data['user_id'] + payload_data = json.loads(base64.urlsafe_b64decode(payload + '===').decode()) + username = payload_data['sub'] exp = payload_data.get('exp', None) self.username = username diff --git a/modcli/pages.py b/modcli/pages.py new file mode 100644 index 0000000..95fb627 --- /dev/null +++ b/modcli/pages.py @@ -0,0 +1,16 @@ +from importlib.resources import files +from string import Template + +_sign_in = Template(files('modcli').joinpath('sign_in.html').read_text(encoding='utf-8')) + + +def signed_in() -> bytes: + return _render('#10B981', 'You\'re signed in', 'Return to the terminal. This window can be closed.') + + +def sign_in_failed() -> bytes: + return _render('#FF3B30', 'Sign-in failed', 'Invalid authentication response.') + + +def _render(color: str, title: str, message: str) -> bytes: + return _sign_in.substitute(color=color, title=title, message=message).encode('utf-8') diff --git a/modcli/settings.py b/modcli/settings.py index 42e984f..81d2a4b 100644 --- a/modcli/settings.py +++ b/modcli/settings.py @@ -2,7 +2,7 @@ CONFIG_DIR = os.path.expanduser('~/.config/modcli') URLS = { - 'labs': ('https://api-labs.moddevices.com/v2', 'https://pipeline-labs.moddevices.com/bundle/'), - 'dev': ('https://api-dev.moddevices.com/v2', 'https://pipeline-dev.moddevices.com/bundle/'), + 'labs': ('https://api-labs.mod.audio/v2', 'https://pipeline-labs.mod.audio/bundle/'), + 'dev': ('https://api-dev.mod.audio/v2', 'https://pipeline-dev.mod.audio/bundle/'), } DEFAULT_ENV = 'labs' diff --git a/modcli/sign_in.html b/modcli/sign_in.html new file mode 100644 index 0000000..520131f --- /dev/null +++ b/modcli/sign_in.html @@ -0,0 +1,56 @@ + + + + + + MOD CLI + + + +
+ + + + + + + + + + + + + + + + + +
MODCLI
+

$title

+

$message

+
+ + diff --git a/modcli/token_receiver.py b/modcli/token_receiver.py new file mode 100644 index 0000000..a990265 --- /dev/null +++ b/modcli/token_receiver.py @@ -0,0 +1,80 @@ +import socket +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer +from urllib import parse + +from modcli import pages + + +def get_open_port(): + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.bind(('', 0)) + s.listen(1) + port = s.getsockname()[1] + s.close() + return port + + +def local_url(port: int): + return f'http://localhost:{port}' + + +def receive_token(port: int, timeout: int, state: str): + '''Return (token, source) for the first token received, either from the + browser redirect to http://localhost:{port} carrying the expected state, + source 'browser', or typed into stdin, source 'paste'. (None, None) on + timeout. Requests without a valid token (a browser's favicon, a missing or + wrong state, e.g. a web page probing localhost) are answered and ignored.''' + received = {} + done = threading.Event() + + def accept(token, source): + received.setdefault('token', token.strip()) + received.setdefault('source', source) + done.set() + + class TokenRequestHandler(BaseHTTPRequestHandler): + def do_HEAD(self): + self.send_response(200) + self.send_header('Content-type', 'text/html') + self.end_headers() + + def do_GET(self): + query = parse.parse_qs(parse.urlsplit(self.path).query) + token = query.get('token', [None])[0] + if token and query.get('state', [None])[0] == state: + accept(token, 'browser') + self._respond(200, pages.signed_in()) + else: + self._respond(400, pages.sign_in_failed()) + + def _respond(self, status_code, content): + self.send_response(status_code) + self.send_header('Content-type', 'text/html; charset=utf-8') + self.end_headers() + self.wfile.write(content) + + def log_message(self, fmt, *args): + pass + + def read_pasted_token(): + try: + token = input() + except EOFError: + return + if token.strip(): + accept(token, 'paste') + + httpd = HTTPServer(('localhost', port), TokenRequestHandler) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + # a pending input() can't be cancelled; the daemon thread is dropped when the command exits + threading.Thread(target=read_pasted_token, daemon=True).start() + try: + done.wait(timeout) + except KeyboardInterrupt: + pass + finally: + httpd.shutdown() + httpd.server_close() + + return received.get('token'), received.get('source') diff --git a/pyproject.toml b/pyproject.toml index 6464440..0da7559 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "mod-devel-cli" -version = "2.0.0" +version = "3.0.0" description = "MOD Command Line Interface" authors = [ {name = "Alex Cunha", email = "alex@mod.audio"} @@ -23,16 +23,26 @@ classifiers = [ "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", ] [project.urls] Homepage = "https://mod.audio/" -[project.optional-dependencies] +[dependency-groups] dev = [ "pylint>=3.0", + "pytest>=8.0", ] [project.scripts] modcli = "modcli.cli:main" +[tool.setuptools.package-data] +modcli = ["*.html"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = ["*_test.py"] + diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..d51fa21 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,7 @@ +import os +import tempfile + +# Importing modcli reads (and creates) ~/.config/modcli, so point the home at a temp dir first +_home = tempfile.mkdtemp() +os.environ['HOME'] = _home +os.environ['USERPROFILE'] = _home diff --git a/tests/auth_test.py b/tests/auth_test.py new file mode 100644 index 0000000..2aad218 --- /dev/null +++ b/tests/auth_test.py @@ -0,0 +1,54 @@ +from unittest import TestCase +from unittest.mock import patch +from urllib import parse + +from modcli import auth +from tests.helpers import FakePost + +API = 'https://api.example.com/v2' +LOCAL_URL = 'http://localhost:1234' + + +class LoginTest(TestCase): + + def test_login(self): + post = FakePost(body='{"message": " the-token "}') + with patch('modcli.http.post', post): + self.assertEqual(auth.login('user', 'secret', API), 'the-token') + url, data = post.calls[0] + self.assertEqual(url, f'{API}/users/tokens') + self.assertEqual((data['user_id'], data['password']), ('user', 'secret')) + + def test_login_error(self): + with patch('modcli.http.post', FakePost(401, '{"error-message": "bad credentials"}')): + with self.assertRaisesRegex(Exception, 'bad credentials'): + auth.login('user', 'wrong', API) + + +class BrowserLoginUrlsTest(TestCase): + + def test_forum_url(self): + base, query = auth.forum_url(API, LOCAL_URL, 'nonce').split('?') + self.assertEqual(base, f'{API}/users/tokens_modforum') + self.assertEqual(parse.parse_qs(query), {'local_url': [LOCAL_URL], 'state': ['nonce']}) + + def test_google_url(self): + base, query = auth.google_url(API, LOCAL_URL, 'nonce').split('?') + self.assertEqual(base, f'{API}/users/tokens_google') + self.assertEqual(parse.parse_qs(query), {'aud': ['api'], 'local_url': [LOCAL_URL], 'state': ['nonce']}) + + +class MagicLinkTest(TestCase): + + def test_send_magic_link(self): + post = FakePost(body='{"success": true, "expires_in": 900}') + with patch('modcli.http.post', post): + self.assertEqual(auth.send_magic_link(API, 'a@b.c', LOCAL_URL, 'nonce'), 900) + self.assertEqual(post.calls, [(f'{API}/users/tokens_magiclink', { + 'email': 'a@b.c', 'aud': 'api', 'local_url': LOCAL_URL, 'state': 'nonce', + })]) + + def test_send_magic_link_error(self): + with patch('modcli.http.post', FakePost(400, '{"error-message": "Email not registered"}')): + with self.assertRaisesRegex(Exception, 'Email not registered'): + auth.send_magic_link(API, 'x@b.c', LOCAL_URL, 'nonce') diff --git a/tests/bundle_test.py b/tests/bundle_test.py new file mode 100644 index 0000000..db79515 --- /dev/null +++ b/tests/bundle_test.py @@ -0,0 +1,47 @@ +import io +import json +import os +import tarfile +import tempfile +from hashlib import md5 +from unittest import TestCase +from unittest.mock import patch + +from modcli import bundle, context +from modcli.http import Response + + +class PublishTest(TestCase): + + def setUp(self): + root = tempfile.mkdtemp() + pkg = os.path.join(root, 'packages', 'my-plugin') + os.makedirs(os.path.join(pkg, 'src')) + for name in ('my-plugin.mk', os.path.join('src', 'plugin.c')): + with open(os.path.join(pkg, name), 'w') as fh: + fh.write(name) + self.project_file = os.path.join(root, 'project.json') + with open(self.project_file, 'w') as fh: + json.dump({'buildroot_pkg': 'my-plugin', 'bundles': [{'name': 'my-plugin.lv2'}]}, fh) + self.calls = [] + + token = patch.object(context.current_env(), 'token', 'the-token') + token.start() + self.addCleanup(token.stop) + + def _post(self, url, data=None, json_data=None, headers=None): + self.calls.append((url, data, json_data)) + if json_data is not None: + return Response(200, json.dumps({'id': 'p1', 'source-href': 'https://bundle/p1/source', 'href': 'https://bundle/p1'})) + return Response(201, f'"{md5(data).hexdigest()}"') + + def test_publish_uploads_the_package(self): + with patch('modcli.http.post', self._post): + bundle.publish(self.project_file, None, force=True) + + (_, _, process), (url, source, _) = self.calls + self.assertEqual(process, {'bundles': [{'name': 'my-plugin.lv2'}], 'rebuild': False}) + self.assertEqual(url, 'https://bundle/p1/source') + with tarfile.open(fileobj=io.BytesIO(source)) as tar: + names = {name.rstrip('/') for name in tar.getnames()} + self.assertLessEqual({'my-plugin/my-plugin.mk', 'my-plugin/src/plugin.c'}, names) diff --git a/tests/cli_test.py b/tests/cli_test.py new file mode 100644 index 0000000..5070c2d --- /dev/null +++ b/tests/cli_test.py @@ -0,0 +1,105 @@ +from unittest import TestCase +from unittest.mock import patch + +from click.testing import CliRunner + +from modcli import cli, config, context +from tests.helpers import TokenHelper + +TOKEN = TokenHelper.generate_token(exp=2000000000, aud='api') +BROWSER_LOGINS = [ + (cli.login_forum, []), + (cli.login_sso, []), + (cli.login_google, []), + (cli.login_magiclink, ['--email', TokenHelper.USER_ID]), +] + + +class BrowserLoginTest(TestCase): + + def setUp(self): + self.opened = [] + self.sent_states = [] + self.expected_states = [] + self.received = (TOKEN, 'browser') + self.timeouts = [] + + def receive_token(port, timeout, state): + self.expected_states.append(state) + self.timeouts.append(timeout) + return self.received + + def send_magic_link(api_url, email, local_url, state): + self.sent_states.append(state) + return 900 + + patches = [ + patch('webbrowser.open', self.opened.append), + patch('modcli.token_receiver.receive_token', receive_token), + patch('modcli.auth.send_magic_link', send_magic_link), + ] + for p in patches: + p.start() + self.addCleanup(p.stop) + + def test_login_stores_token(self): + for command, args in BROWSER_LOGINS: + with self.subTest(command=command.name): + result = CliRunner().invoke(command, args) + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn(f'You\'re now logged in as [{TokenHelper.USER_ID}]', result.output) + self.assertEqual(context.active_token(), TOKEN) + + def test_login_requires_the_state_it_sent(self): + for command, args in BROWSER_LOGINS: + with self.subTest(command=command.name): + CliRunner().invoke(command, args) + state = self.expected_states[-1] + self.assertTrue(state) + sent = self.sent_states[-1] if command is cli.login_magiclink else self.opened[-1] + self.assertIn(state, sent) + + def test_login_opens_the_sign_in_url(self): + CliRunner().invoke(cli.login_google, []) + self.assertEqual(len(self.opened), 1) + self.assertIn('aud=api', self.opened[0]) + + def test_magiclink_waits_as_long_as_the_link_lives(self): + CliRunner().invoke(cli.login_magiclink, ['--email', TokenHelper.USER_ID]) + self.assertEqual(self.timeouts[-1], 900) + + def test_one_time_prints_token_only(self): + result = CliRunner().invoke(cli.login_google, ['-o']) + self.assertEqual(result.exit_code, 0) + self.assertTrue(result.output.strip().endswith(TOKEN)) + + def test_failed_login_exits_with_error(self): + self.received = (None, None) + result = CliRunner().invoke(cli.login_google, []) + self.assertEqual(result.exit_code, 1) + self.assertIn('Authentication failed!', result.output) + + +class ConfirmAllTest(TestCase): + + def test_confirm_all_is_accepted_and_ignored(self): + for command, args in [(cli.login, ['-u', 'user', '-p', 'secret']), *BROWSER_LOGINS]: + with self.subTest(command=command.name): + params = command.make_context(command.name, ['-y', *args]).params + self.assertNotIn('confirm_all', params) + + +class SetTokenTest(TestCase): + + def test_set_token_decodes_base64url_payload(self): + # A payload whose base64url encoding uses '-' or '_', which plain base64 decoding garbles + user_id = next(f'user{i}?>' for i in range(1000) + if set('-_') & set(TokenHelper.generate_token(f'user{i}?>').split('.')[1])) + env = config.EnvSettings('test', 'https://api', 'https://bundle') + env.set_token(TokenHelper.generate_token(user_id)) + self.assertEqual(env.username, user_id) + + def test_set_token_reads_sub_and_exp(self): + env = config.EnvSettings('test', 'https://api', 'https://bundle') + env.set_token(TokenHelper.generate_token('dev@partner.com', exp=2000000000, aud='api', holder='person')) + self.assertEqual((env.username, env.exp), ('dev@partner.com', 2000000000)) diff --git a/tests/config_test.py b/tests/config_test.py new file mode 100644 index 0000000..24485f8 --- /dev/null +++ b/tests/config_test.py @@ -0,0 +1,43 @@ +import os +import stat +import sys +import tempfile +from unittest import TestCase, skipIf + +from modcli import config +from tests.helpers import TokenHelper + +TOKEN = TokenHelper.generate_token(exp=2000000000) + + +class CliContextTest(TestCase): + + def setUp(self): + self.path = os.path.join(tempfile.mkdtemp(), 'modcli') + context = config.CliContext(self.path) + context.add_env('test', 'https://api.example.com/v2/', 'https://bundle.example.com/') + context.set_active_env('test') + context.current_env().set_token(TOKEN) + context.save() + + def _file(self, name): + return os.path.join(self.path, name) + + def test_save_and_read(self): + env = config.CliContext.read(self.path).current_env() + self.assertEqual((env.name, env.api_url, env.bundle_url), ('test', 'https://api.example.com/v2', 'https://bundle.example.com')) + self.assertEqual((env.token, env.username, env.exp), (TOKEN, TokenHelper.USER_ID, 2000000000)) + with open(self._file('access_token')) as fh: + self.assertEqual(fh.read().strip(), TOKEN) + + @skipIf(sys.platform == 'win32', 'POSIX permissions') + def test_files_are_private(self): + for name in ('context.json', 'access_token'): + self.assertEqual(stat.S_IMODE(os.stat(self._file(name)).st_mode), 0o600) + + def test_clear(self): + context = config.CliContext.read(self.path) + context.clear() + self.assertFalse(os.path.exists(self._file('context.json'))) + self.assertFalse(os.path.exists(self._file('access_token'))) + self.assertEqual(config.CliContext.read(self.path).environments, {}) diff --git a/tests/helpers/__init__.py b/tests/helpers/__init__.py new file mode 100644 index 0000000..b1d2153 --- /dev/null +++ b/tests/helpers/__init__.py @@ -0,0 +1,2 @@ +from tests.helpers.token_helper import TokenHelper +from tests.helpers.http_helper import FakePost diff --git a/tests/helpers/http_helper.py b/tests/helpers/http_helper.py new file mode 100644 index 0000000..8180b4d --- /dev/null +++ b/tests/helpers/http_helper.py @@ -0,0 +1,13 @@ +from modcli.http import Response + + +class FakePost: + '''Stands in for modcli.http.post: records each call and answers with `response`.''' + + def __init__(self, status_code=200, body='{"message": "the-token"}'): + self.response = Response(status_code, body) + self.calls = [] + + def __call__(self, url, json_data=None, **_): + self.calls.append((url, json_data)) + return self.response diff --git a/tests/helpers/token_helper.py b/tests/helpers/token_helper.py new file mode 100644 index 0000000..f6f52fb --- /dev/null +++ b/tests/helpers/token_helper.py @@ -0,0 +1,14 @@ +import base64 +import json + + +class TokenHelper: + USER_ID = 'dev@partner.com' + + @staticmethod + def generate_token(user_id=USER_ID, **payload): + # Unsigned: the CLI only decodes the payload, it never verifies the signature + def encode(data: dict) -> str: + return base64.urlsafe_b64encode(json.dumps(data).encode()).decode().rstrip('=') + header = encode({'alg': 'EdDSA', 'typ': 'JWT'}) + return f'{header}.{encode({"sub": user_id, **payload})}.signature' diff --git a/tests/token_receiver_test.py b/tests/token_receiver_test.py new file mode 100644 index 0000000..eccb696 --- /dev/null +++ b/tests/token_receiver_test.py @@ -0,0 +1,78 @@ +import threading +import time +import urllib.request +from unittest import TestCase +from unittest.mock import patch +from urllib.error import HTTPError + +from modcli import pages, token_receiver + + +def _get(port: int, path: str) -> int: + # 127.0.0.1, not localhost: Windows tries ::1 first and spends ~2s on each refused connection + try: + with urllib.request.urlopen(f'http://127.0.0.1:{port}{path}') as response: + return response.status + except HTTPError as e: + return e.code + + +class ReceiveTokenTest(TestCase): + + def _receive(self, paths, state='s1', pasted=None, timeout=3): + '''Run receive_token while a thread sends the browser requests in `paths`. + `pasted` is what the user types; None leaves stdin empty.''' + def fake_input(): + if pasted is None: + raise EOFError + time.sleep(0.3) + return pasted + + port = token_receiver.get_open_port() + statuses = [] + + def browser(): + time.sleep(0.5) + statuses.extend(_get(port, path) for path in paths) + + thread = threading.Thread(target=browser, daemon=True) + thread.start() + with patch('builtins.input', fake_input): + result = token_receiver.receive_token(port, timeout, state) + thread.join(timeout=1) + return result, statuses + + def test_browser_token_with_matching_state(self): + result, statuses = self._receive(['/?token=good&state=s1']) + self.assertEqual(result, ('good', 'browser')) + self.assertEqual(statuses, [200]) + + def test_invalid_requests_are_ignored(self): + paths = ['/favicon.ico', '/?state=s1', '/?token=bad&state=wrong', '/?token=bad', '/?token=good&state=s1'] + result, statuses = self._receive(paths) + self.assertEqual(result, ('good', 'browser')) + self.assertEqual(statuses, [400, 400, 400, 400, 200]) + + def test_pasted_token(self): + result, _ = self._receive([], pasted=' pasted-token \n') + self.assertEqual(result, ('pasted-token', 'paste')) + + def test_blank_paste_is_ignored(self): + result, _ = self._receive(['/?token=good&state=s1'], pasted=' ') + self.assertEqual(result, ('good', 'browser')) + + def test_timeout(self): + result, _ = self._receive([], timeout=1) + self.assertEqual(result, (None, None)) + + def test_local_url(self): + self.assertEqual(token_receiver.local_url(1234), 'http://localhost:1234') + + +class PagesTest(TestCase): + + def test_pages_render(self): + for page in (pages.signed_in(), pages.sign_in_failed()): + html = page.decode('utf-8') + self.assertTrue(html.startswith('')) + self.assertNotIn('$', html) diff --git a/uv.lock b/uv.lock index f941593..128fd69 100644 --- a/uv.lock +++ b/uv.lock @@ -49,6 +49,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/50/3d/9373ad9c56321fdab5b41197068e1d8c25883b3fea29dd361f9b55116869/dill-0.4.0-py3-none-any.whl", hash = "sha256:44f54bf6412c2c8464c14e8243eb163690a9800dbe2c367330883b19c7561049", size = 119668, upload-time = "2025-04-16T00:41:47.671Z" }, ] +[[package]] +name = "exceptiongroup" +version = "1.3.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8a/0e/97c33bf5009bdbac74fd2beace167cab3f978feb69cc36f1ef79360d6c4e/exceptiongroup-1.3.1-py3-none-any.whl", hash = "sha256:a7a39a3bd276781e98394987d3a5701d0c4edffb633bb7a5144577f82c773598", size = 16740, upload-time = "2025-11-21T23:01:53.443Z" }, +] + +[[package]] +name = "iniconfig" +version = "2.3.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/01/e1/2069291243c926a2ff1cd706c7f3eeb9b62144bf60f77c9fb9ff2fb26bd3/iniconfig-2.3.1.tar.gz", hash = "sha256:67f4b9c50da0dedf52af349e7749a80a9057a5031199791b906c3bb3ae878960", size = 21209, upload-time = "2026-10-06T22:48:38.076Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/56/43/4ca9e49d27a1fcf6bece6f6aec0ea46bb9112489b93d4b688fb415457bdb/iniconfig-2.3.1-py3-none-any.whl", hash = "sha256:9121e2c1fdb355232495be3194c8dfe87ccc2d5dee45947b78e68f499790d7a7", size = 7552, upload-time = "2026-10-06T22:48:36.959Z" }, +] + [[package]] name = "isort" version = "7.0.0" @@ -69,23 +90,35 @@ wheels = [ [[package]] name = "mod-devel-cli" -version = "2.0.0" +version = "3.0.0" source = { editable = "." } dependencies = [ { name = "click" }, ] -[package.optional-dependencies] +[package.dev-dependencies] dev = [ { name = "pylint" }, + { name = "pytest" }, ] [package.metadata] -requires-dist = [ - { name = "click", specifier = ">=8.0" }, - { name = "pylint", marker = "extra == 'dev'", specifier = ">=3.0" }, +requires-dist = [{ name = "click", specifier = ">=8.0" }] + +[package.metadata.requires-dev] +dev = [ + { name = "pylint", specifier = ">=3.0" }, + { name = "pytest", specifier = ">=8.0" }, +] + +[[package]] +name = "packaging" +version = "26.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/7d/fa/3944b40b07da9ce895c0e6303a5ab7d53da063554f534556b134a54d6093/packaging-26.3.tar.gz", hash = "sha256:94edc256424af38762eb31306eed28beb9f0efc50a8837492c9d6fd6004aed79", size = 313412, upload-time = "2026-08-04T18:15:28.737Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/63/34/ba1c580383c9eada3711951fef0795c80b829a078d72188184bcab9dd527/packaging-26.3-py3-none-any.whl", hash = "sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c", size = 129956, upload-time = "2026-08-04T18:15:27.159Z" }, ] -provides-extras = ["dev"] [[package]] name = "platformdirs" @@ -96,6 +129,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/28/3bfe2fa5a7b9c46fe7e13c97bda14c895fb10fa2ebf1d0abb90e0cea7ee1/platformdirs-4.5.1-py3-none-any.whl", hash = "sha256:d03afa3963c806a9bed9d5125c8f4cb2fdaf74a55ab60e5d59b3fde758104d31", size = 18731, upload-time = "2025-12-05T13:52:56.823Z" }, ] +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + +[[package]] +name = "pygments" +version = "2.21.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/49/2e/ced460408999b33da6b31b0021b0f37d329e202d4169aeb164493778f25b/pygments-2.21.0.tar.gz", hash = "sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c", size = 5005329, upload-time = "2026-08-17T08:02:48.824Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/71/46/17f022dd3e953bf20a04a028a21ec746d942f8d2af30fa0f124fa0e6a684/pygments-2.21.0-py3-none-any.whl", hash = "sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9", size = 1250147, upload-time = "2026-08-17T08:02:44.912Z" }, +] + [[package]] name = "pylint" version = "4.0.4" @@ -115,6 +166,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a6/92/d40f5d937517cc489ad848fc4414ecccc7592e4686b9071e09e64f5e378e/pylint-4.0.4-py3-none-any.whl", hash = "sha256:63e06a37d5922555ee2c20963eb42559918c20bd2b21244e4ef426e7c43b92e0", size = 536425, upload-time = "2025-11-30T13:29:02.53Z" }, ] +[[package]] +name = "pytest" +version = "9.1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "exceptiongroup", marker = "python_full_version < '3.11'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, + { name = "tomli", marker = "python_full_version < '3.11'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" }, +] + [[package]] name = "tomli" version = "2.3.0"