diff --git a/src/cryptoadvance/specter/cli/cli_server.py b/src/cryptoadvance/specter/cli/cli_server.py index bf3d4c9c1..44e57276a 100644 --- a/src/cryptoadvance/specter/cli/cli_server.py +++ b/src/cryptoadvance/specter/cli/cli_server.py @@ -179,7 +179,14 @@ def server( print( " * Running in HWI Bridge mode.\n" " * You can configure access to the API " - "at: %s://%s:%d/hwi/settings" % ("http", host, app.config["PORT"]) + "at: %s://%s:%d%s%s/hwi/settings" + % ( + "http", + host, + app.config["PORT"], + app.config["APP_URL_PREFIX"], + app.config["SPECTER_URL_PREFIX"], + ) ) # debug is false by default diff --git a/src/cryptoadvance/specter/hwi_server.py b/src/cryptoadvance/specter/hwi_server.py index 59f0e23f4..43833cc57 100644 --- a/src/cryptoadvance/specter/hwi_server.py +++ b/src/cryptoadvance/specter/hwi_server.py @@ -1,7 +1,16 @@ import json, os, random, requests -from flask import Blueprint, Flask, jsonify, url_for, redirect, render_template, request +from flask import ( + Blueprint, + Flask, + jsonify, + redirect, + render_template, + request, + url_for, +) from .server_endpoints import flash from flask import current_app as app +from flask_login import current_user, login_required from flask_cors import CORS from .hwi_rpc import HWIBridge from .helpers import deep_update, hwi_get_config, save_hwi_bridge_config @@ -10,13 +19,14 @@ import logging logger = logging.getLogger(__name__) hwi_server = Blueprint("hwi_server", __name__) +hwi_server_settings = Blueprint("hwi_server_settings", __name__) CORS(hwi_server) rand = random.randint(0, int(1e32)) # to force style refresh @hwi_server.route("/", methods=["GET"]) def index(): - return redirect(url_for("hwi_server.hwi_bridge_settings")) + return redirect(url_for("hwi_server_settings.hwi_bridge_settings")) @hwi_server.route("/api/", methods=["POST"]) @@ -100,8 +110,15 @@ def api(): return jsonify(app.specter.hwi.jsonrpc(data)) -@hwi_server.route("/settings/", methods=["GET", "POST"]) +@hwi_server_settings.route("/settings/", methods=["GET", "POST"]) +@login_required def hwi_bridge_settings(): + if app.config.get("LOGIN_DISABLED") and ( + not current_user.is_authenticated or not current_user.is_admin + ): + app.login("admin") + if not current_user.is_admin: + return "Forbidden", 403 config = hwi_get_config(app.specter) if request.method == "POST": action = request.form["action"] diff --git a/src/cryptoadvance/specter/server.py b/src/cryptoadvance/specter/server.py index a94233902..a2f742a6e 100644 --- a/src/cryptoadvance/specter/server.py +++ b/src/cryptoadvance/specter/server.py @@ -23,7 +23,7 @@ from werkzeug.wrappers import Response from cryptoadvance.specter.hwi_rpc import HWIBridge from .htmlsafebabel import HTMLSafeBabel -from .hwi_server import hwi_server +from .hwi_server import hwi_server, hwi_server_settings from .services.callbacks import after_serverpy_init_app, specter_added_to_flask_app from .specter import Specter from .util.specter_migrator import SpecterMigrator @@ -229,8 +229,16 @@ def init_app(app: SpecterFlask, hwibridge=False, specter=None): app.logger.info("Login enabled") app.config["LOGIN_DISABLED"] = False app.logger.info("Initializing Controller ...") + hwi_settings_prefix = f"{app.config['SPECTER_URL_PREFIX']}/hwi" + app.register_blueprint(hwi_server_settings, url_prefix=hwi_settings_prefix) app.register_blueprint(hwi_server, url_prefix="/hwi") - csrf.exempt(hwi_server) + csrf.exempt(app.view_functions["hwi_server.api"]) + if hwi_settings_prefix != "/hwi": + app.add_url_rule( + "/hwi/settings/", + "hwi_server.hwi_bridge_settings", + lambda: redirect(url_for("hwi_server_settings.hwi_bridge_settings")), + ) if not hwibridge: with app.app_context(): from cryptoadvance.specter.server_endpoints import controller @@ -258,10 +266,17 @@ def init_app(app: SpecterFlask, hwibridge=False, specter=None): importlib.reload(controller) importlib.reload(serviceController) else: + with app.app_context(): + from cryptoadvance.specter.server_endpoints.auth import auth_endpoint + + app.register_blueprint( + auth_endpoint, + url_prefix=f"{app.config['SPECTER_URL_PREFIX']}/auth", + ) @app.route("/", methods=["GET"]) def index(): - return redirect(url_for("hwi_server.hwi_bridge_settings")) + return redirect(url_for("hwi_server_settings.hwi_bridge_settings")) if app.config["SPECTER_API_ACTIVE"]: app.logger.info("Initializing REST ...") diff --git a/tests/test_hwi_server.py b/tests/test_hwi_server.py new file mode 100644 index 000000000..e9386956e --- /dev/null +++ b/tests/test_hwi_server.py @@ -0,0 +1,180 @@ +import re +import sys +from uuid import uuid4 + +import pytest + +from cryptoadvance.specter.config import TestConfig +from cryptoadvance.specter.helpers import hwi_get_config +from cryptoadvance.specter.server import create_app, init_app +from cryptoadvance.specter.specter import Specter + + +def make_scoped_app(tmp_path, hwibridge, auth_method="none"): + config_name = f"ScopedTestConfig_{uuid4().hex}" + config_class = type( + config_name, + (TestConfig,), + { + "__module__": __name__, + "SPECTER_DATA_FOLDER": str(tmp_path), + "SPECTER_URL_PREFIX": "/spc", + "SESSION_COOKIE_PATH": "/spc", + "SESSION_PROTECTION": None, + "SKIP_HWI_INITIALISATION_AT_STARTUP": True, + "SPECTER_API_ACTIVE": False, + }, + ) + setattr(sys.modules[__name__], config_name, config_class) + specter = Specter(data_folder=str(tmp_path), checker_threads=False) + if auth_method != "none": + specter.update_auth(auth_method, 10, 1) + + app = create_app(config_class) + app.config["TESTING"] = True + with app.app_context(): + init_app(app, hwibridge=hwibridge, specter=specter) + return app + + +def csrf_token(response): + match = re.search(b'name="csrf_token" value="([^"]+)', response.data) + assert match is not None + return match.group(1).decode() + + +@pytest.mark.parametrize("hwibridge", [False, True]) +def test_hwi_settings_use_session_scoped_route(tmp_path, hwibridge): + app = make_scoped_app(tmp_path, hwibridge) + client = app.test_client() + + legacy_response = client.get("/hwi/settings/") + assert legacy_response.status_code == 302 + assert legacy_response.headers["Location"] == "/spc/hwi/settings/" + + settings_response = client.get("/spc/hwi/settings/") + assert settings_response.status_code == 200 + assert "Path=/spc" in settings_response.headers["Set-Cookie"] + + missing_csrf_response = client.post( + "/spc/hwi/settings/", + data={"action": "update", "whitelisted_domains": "http://attacker/"}, + ) + assert missing_csrf_response.status_code in (302, 400) + assert hwi_get_config(app.specter)["whitelisted_domains"] != "http://attacker/" + + invalid_csrf_response = client.post( + "/spc/hwi/settings/", + data={ + "action": "update", + "csrf_token": "invalid", + "whitelisted_domains": "http://attacker/", + }, + ) + assert invalid_csrf_response.status_code in (302, 400) + assert hwi_get_config(app.specter)["whitelisted_domains"] != "http://attacker/" + + update_response = client.post( + "/spc/hwi/settings/", + data={ + "action": "update", + "csrf_token": csrf_token(settings_response), + "whitelisted_domains": "http://example.com/", + }, + ) + assert update_response.status_code == 200 + assert ( + hwi_get_config(app.specter)["whitelisted_domains"].strip() + == "http://example.com/" + ) + + +@pytest.mark.parametrize("hwibridge", [False, True]) +def test_hwi_settings_require_authenticated_admin(tmp_path, hwibridge): + app = make_scoped_app(tmp_path, hwibridge=hwibridge, auth_method="usernamepassword") + client = app.test_client() + + anonymous_response = client.get("/spc/hwi/settings/") + assert anonymous_response.status_code == 302 + assert anonymous_response.headers["Location"].startswith( + "/spc/auth/login?next=%2Fspc%2Fhwi%2Fsettings%2F" + ) + assert client.get("/spc/auth/login").status_code == 200 + + login_response = client.post( + "/spc/auth/login", + data={ + "username": "admin", + "password": "admin", + "next": "/spc/hwi/settings/", + }, + ) + assert login_response.status_code == 302 + assert login_response.headers["Location"] == "/spc/hwi/settings/" + assert client.get(login_response.headers["Location"]).status_code == 200 + + app.specter.user_manager.create_user( + user_id="nonadmin", + username="nonadmin", + plaintext_password="nonadmin", + config={}, + ) + nonadmin_client = app.test_client() + nonadmin_login_response = nonadmin_client.post( + "/spc/auth/login", + data={ + "username": "nonadmin", + "password": "nonadmin", + "next": "/spc/hwi/settings/", + }, + ) + assert nonadmin_login_response.status_code == 302 + assert nonadmin_client.get("/spc/hwi/settings/").status_code == 403 + + +def test_hwi_settings_force_admin_when_login_disabled(tmp_path): + app = make_scoped_app(tmp_path, hwibridge=True, auth_method="usernamepassword") + app.specter.user_manager.create_user( + user_id="nonadmin", + username="nonadmin", + plaintext_password="nonadmin", + config={}, + ) + client = app.test_client() + login_response = client.post( + "/spc/auth/login", + data={ + "username": "nonadmin", + "password": "nonadmin", + "next": "/spc/hwi/settings/", + }, + ) + assert login_response.status_code == 302 + + app.specter.update_auth("none", 10, 1) + app.config["LOGIN_DISABLED"] = True + assert client.get("/spc/hwi/settings/").status_code == 200 + with client.session_transaction(path="/spc/hwi/settings/") as session: + assert session["_user_id"] == "admin" + + +@pytest.mark.parametrize("hwibridge", [False, True]) +def test_hwi_api_remains_csrf_exempt(tmp_path, hwibridge): + app = make_scoped_app(tmp_path, hwibridge=hwibridge) + app.specter.hwi.exposed_rpc["enumerate"] = lambda **kwargs: [] + client = app.test_client() + client.environ_base["HTTP_ORIGIN"] = "http://127.0.0.1:25441/" + + response = client.post( + "/hwi/api/", + json={ + "jsonrpc": "2.0", + "method": "enumerate", + "id": 1, + "params": {}, + "forwarded_request": True, + }, + ) + + assert response.status_code == 200 + assert response.get_json() == {"id": 1, "jsonrpc": "2.0", "result": []}