diff --git a/backend/.env.example b/backend/.env.example index 7f6474d..968a71f 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -37,7 +37,10 @@ T_PLUS_CONNECTION_FILE= # 必须填写在客户 SQL Server 2008 R2 上验证过的驱动名称。 # 本地 PoC 可复用已安装的 SQL Server 驱动；不自动安装或升级客户驱动。 T_PLUS_ODBC_DRIVER= -T_PLUS_ENCRYPT=false +# 默认启用传输加密。旧实例 TLS 不兼容时，仅隔离开发环境可显式设为 false， +# 记录原因并重新验收；程序不会自动降级。真实环境值后续按验收结果填写。 +T_PLUS_ENCRYPT=true +# 内网自签证书可暂用 true；它不校验证书身份。部署可信证书后改为 false。 T_PLUS_TRUST_SERVER_CERTIFICATE=true T_PLUS_CONNECT_TIMEOUT=10 T_PLUS_QUERY_TIMEOUT=120 diff --git a/backend/app/core/config.py b/backend/app/core/config.py index fcd6842..b8cac5d 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -69,7 +69,7 @@ class Settings(BaseSettings): t_plus_connection_file: str = "" # 部署前必须指定已在 SQL Server 2008 R2 实测的驱动，不自动选择最新版。 t_plus_odbc_driver: str = "" - t_plus_encrypt: bool = False + t_plus_encrypt: bool = True t_plus_trust_server_certificate: bool = True t_plus_connect_timeout: int = 10 t_plus_query_timeout: int = Field(default=120, ge=1, le=600) @@ -115,6 +115,7 @@ class Settings(BaseSettings): 连接 URL 只在启用 T+ 适配器时使用；密码由 SQLAlchemy URL 对象保存， 日志中调用 ``render_as_string()`` 时仍会默认隐藏密码。 + ODBC 的分号和花括号由方言统一转义，调用方不得再手工拼接连接串。 """ query = { "driver": self.t_plus_odbc_driver, diff --git a/backend/tests/unit/test_tplus_compatibility_script.py b/backend/tests/unit/test_tplus_compatibility_script.py new file mode 100644 index 0000000..cc35ed9 --- /dev/null +++ b/backend/tests/unit/test_tplus_compatibility_script.py @@ -0,0 +1,92 @@ +"""兼容性脚本与应用使用相同连接策略，失败报告不泄露连接数据。""" + +import importlib.util +import json +import sys +from datetime import datetime +from decimal import Decimal +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from app.core.config import Settings + + +@pytest.fixture +def script(monkeypatch): + # 纯单元测试不需要安装 ODBC 驱动或连接源库。 + monkeypatch.setitem(sys.modules, "pyodbc", MagicMock()) + path = Path(__file__).resolve().parents[3] / "scripts/tplus/check_compatibility.py" + spec = importlib.util.spec_from_file_location("compatibility_script", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + settings = Settings(_env_file=None, app_secret_key="x" * 32, t_plus_encrypt=True) + monkeypatch.setattr(module, "get_settings", lambda: settings) + return module + + +def test_脚本复用连接文件校验与加密设置(script, tmp_path): + path = tmp_path / "reader.json" + payload = { + "host": "localhost", + "port": 1433, + "database": "data;x}", + "user": "reader;other}", + "password": "test-only;x}", + "driver": "TestDriver", + } + path.write_text(json.dumps(payload), encoding="utf-8") + resolved = script.compatibility_settings(path) + assert resolved.t_plus_database_url.database == payload["database"] + assert resolved.t_plus_database_url.username == payload["user"] + assert resolved.t_plus_database_url.query["Encrypt"] == "yes" + assert script.get_settings().t_plus_enabled is False + payload["user"] = "sa" + path.write_text(json.dumps(payload), encoding="utf-8") + with pytest.raises(RuntimeError, match="专用只读"): + script.compatibility_settings(path) + + +@pytest.mark.parametrize( + "version,passed", + [ + ("10.50.1600", True), + ("10.50.1600.0", True), + ("10.50.6000", False), + ("16.0.4165.4", False), + ], +) +def test_同主版本或2022副本不能冒充备份源同版本(script, version, passed): + connection = MagicMock() + cursor = connection.cursor.return_value + cursor.execute.return_value.fetchone.side_effect = [ + (version,), + ("中文核验", Decimal("12345.6789"), datetime(2008, 1, 2, 3, 4, 5)), + (0, 0, 0), + ] + connection.getinfo.return_value = "test-driver" + report = script._inspect_connection(connection, "10.50.1600") + assert report["compatible"] is passed + cursor.close.assert_called_once() + + +def test_连接失败不降级且报告不带驱动异常原文(script, tmp_path, monkeypatch, capsys): + report_path = tmp_path / "report.json" + monkeypatch.setattr( + sys, "argv", ["check", "--connection-file", "unused.json", "--report", str(report_path)] + ) + monkeypatch.setattr(script, "compatibility_settings", lambda path: script.get_settings()) + engine = MagicMock() + engine.raw_connection.side_effect = RuntimeError("PWD=secret-test-only;手机号") + create = MagicMock(return_value=engine) + monkeypatch.setattr(script, "create_engine", create) + assert script.main() == 2 + create.assert_called_once() + engine.raw_connection.assert_called_once() + engine.dispose.assert_called_once() + assert json.loads(report_path.read_text(encoding="utf-8")) == { + "compatible": False, + "error_type": "RuntimeError", + } + assert "secret-test-only" not in capsys.readouterr().out diff --git a/backend/tests/unit/test_tplus_config.py b/backend/tests/unit/test_tplus_config.py index 58a8f90..016690a 100644 --- a/backend/tests/unit/test_tplus_config.py +++ b/backend/tests/unit/test_tplus_config.py @@ -3,6 +3,7 @@ from pathlib import Path import pytest +from sqlalchemy.dialects.mssql.pyodbc import MSDialect_pyodbc from app.core.config import Settings from app.services.tplus.mapping import MappingStatus, get_mapping @@ -33,6 +34,34 @@ def test_T加未启用时读取器不建立连接() -> None: assert settings.t_plus_enabled is False +def test_T加默认加密且允许显式隔离环境回退(monkeypatch) -> None: + monkeypatch.delenv("T_PLUS_ENCRYPT", raising=False) + monkeypatch.delenv("T_PLUS_TRUST_SERVER_CERTIFICATE", raising=False) + assert _settings().t_plus_database_url.query["Encrypt"] == "yes" + assert _settings().t_plus_database_url.query["TrustServerCertificate"] == "yes" + url = _settings(t_plus_encrypt=False, t_plus_trust_server_certificate=False).t_plus_database_url + assert url.query["Encrypt"] == "no" + assert url.query["TrustServerCertificate"] == "no" + + +@pytest.mark.parametrize("value", ["a;b", "a}b", "{prefix", "中文;UID=other}x"]) +def test_ODBC特殊字符保持字段边界且不泄露密码(value) -> None: + settings = _settings( + t_plus_odbc_driver="Driver}Name", + t_plus_database=value, + t_plus_user=value, + t_plus_password=value, + t_plus_encrypt=True, + ) + url = settings.t_plus_database_url + args, _ = MSDialect_pyodbc().create_connect_args(url) + escaped = "{" + value.replace("}", "}}") + "}" + assert f"Database={escaped};UID={escaped};PWD={escaped};" in args[0] + assert "DRIVER={Driver}}Name}" in args[0] + assert ";Encrypt=yes;" in args[0] + assert url.render_as_string().split("@", 1)[0].endswith(":***") + + def test_T加查询文件覆盖六类实体() -> None: reader = TPlusReader.__new__(TPlusReader) reader.query_dir = Path(__file__).resolve().parents[3] / "db" / "queries" / "tplus" diff --git a/scripts/tplus/check_compatibility.py b/scripts/tplus/check_compatibility.py index df7c29c..f42cb5e 100644 --- a/scripts/tplus/check_compatibility.py +++ b/scripts/tplus/check_compatibility.py @@ -1,4 +1,4 @@ -"""只读核验真实 SQL Server 2008 R2；2022 PoC 不能获得兼容通过结果。""" +"""只读核验备份源同版本 SQL Server；2022 PoC 不能获得兼容通过结果。""" import argparse import json @@ -8,41 +8,36 @@ from decimal import Decimal from pathlib import Path import pyodbc +from sqlalchemy import create_engine +from sqlalchemy.orm import Session sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "backend")) -from sqlalchemy import URL, create_engine -from sqlalchemy.orm import Session - +from app.core.config import Settings, get_settings +from app.core.tplus_db import resolve_tplus_settings from app.services.tplus.mapping import DETAIL_MAPPINGS, TPLUS_MAPPINGS from app.services.tplus.reader import TPlusReader -def main() -> int: - """检查版本、中文、金额、日期和写权限，结果不包含凭据或客户记录。""" - parser = argparse.ArgumentParser(description="SQL Server 2008 R2 兼容性验收") - parser.add_argument("--connection-file", required=True, type=Path) - parser.add_argument("--report", required=True, type=Path) - args = parser.parse_args() - config = json.loads(args.connection_file.read_text(encoding="utf-8-sig")) - connection_string = ( - "DRIVER={" - + config["driver"] - + "};SERVER=" - + config["host"] - + "," - + str(config["port"]) - + ";DATABASE=" - + config["database"] - + ";UID=" - + config["user"] - + ";PWD={" - + config["password"].replace("}", "}}") - + "};" +def compatibility_settings(path: Path) -> Settings: + """复用应用的连接文件校验、URL 转义和 TLS 配置，不修改全局设置。""" + settings = get_settings().model_copy( + update={"t_plus_enabled": True, "t_plus_connection_file": str(path)} ) - with pyodbc.connect(connection_string, timeout=10) as connection: - connection.timeout = 15 - cursor = connection.cursor() + return resolve_tplus_settings(settings) + + +def _version_matches(actual: str, expected: str) -> bool: + def parts(value): + values = value.split(".") + return tuple(values + ["0"] * (4 - len(values))) + + return parts(actual) == parts(expected) + + +def _inspect_connection(connection, expected_version: str) -> dict: + cursor = connection.cursor() + try: version = str( cursor.execute( "SELECT CAST(SERVERPROPERTY('ProductVersion') AS varchar(80))" @@ -57,45 +52,70 @@ def main() -> int: "HAS_PERMS_BY_NAME(DB_NAME(),'DATABASE','UPDATE'), " "HAS_PERMS_BY_NAME(DB_NAME(),'DATABASE','DELETE')" ).fetchone() - checks = { - "sqlserver_2008r2": version.startswith("10.50."), - "unicode_roundtrip": values[0] == "中文核验", - "decimal_exact": values[1] == Decimal("12345.6789"), - "datetime_exact": values[2] == datetime(2008, 1, 2, 3, 4, 5), - "no_dml_permissions": all(x == 0 for x in permissions), - } - report = { - "server_version": version, - "driver_name": connection.getinfo(pyodbc.SQL_DRIVER_NAME), - "driver_version": connection.getinfo(pyodbc.SQL_DRIVER_VER), - "checks": checks, - "compatible": all(checks.values()), - } - url = URL.create( - "mssql+pyodbc", - username=config["user"], - password=config["password"], - host=config["host"], - port=config["port"], - database=config["database"], - query={"driver": config["driver"]}, - ) - engine = create_engine(url, hide_parameters=True) - adapter_checks = {} - try: - with Session(engine) as session: - reader = TPlusReader(session) - for mapping in (*TPLUS_MAPPINGS, *DETAIL_MAPPINGS): - try: - reader.fetch_entity(mapping.entity_type, max_rows=1) - adapter_checks[mapping.entity_type] = True - except Exception as exc: - adapter_checks[mapping.entity_type] = False - report.setdefault("errors", {})[mapping.entity_type] = type(exc).__name__ finally: - engine.dispose() - report["adapter_queries"] = adapter_checks - report["compatible"] = report["compatible"] and all(adapter_checks.values()) + cursor.close() + checks = { + "sqlserver_2008r2": version.startswith("10.50."), + "backup_source_version": _version_matches(version, expected_version), + "unicode_roundtrip": values[0] == "中文核验", + "decimal_exact": values[1] == Decimal("12345.6789"), + "datetime_exact": values[2] == datetime(2008, 1, 2, 3, 4, 5), + "no_dml_permissions": all(x == 0 for x in permissions), + } + return { + "server_version": version, + "expected_version": expected_version, + "driver_name": connection.getinfo(pyodbc.SQL_DRIVER_NAME), + "driver_version": connection.getinfo(pyodbc.SQL_DRIVER_VER), + "checks": checks, + "compatible": all(checks.values()), + } + + +def _inspect_adapter(engine, report: dict) -> None: + checks = {} + with Session(engine) as session: + reader = TPlusReader(session) + for mapping in (*TPLUS_MAPPINGS, *DETAIL_MAPPINGS): + try: + reader.fetch_entity(mapping.entity_type, max_rows=1) + checks[mapping.entity_type] = True + except Exception as exc: + session.rollback() + checks[mapping.entity_type] = False + report.setdefault("errors", {})[mapping.entity_type] = type(exc).__name__ + report["adapter_queries"] = checks + report["compatible"] = report["compatible"] and all(checks.values()) + + +def main() -> int: + """检查版本、中文、金额、日期和写权限，报告不包含凭据或客户记录。""" + parser = argparse.ArgumentParser(description="SQL Server 2008 R2 兼容性验收") + parser.add_argument("--connection-file", required=True, type=Path) + parser.add_argument("--report", required=True, type=Path) + parser.add_argument("--expected-version", default="10.50.1600", help="备份源版本基准") + args = parser.parse_args() + engine = None + try: + settings = compatibility_settings(args.connection_file) + engine = create_engine( + settings.t_plus_database_url, + hide_parameters=True, + connect_args={"timeout": settings.t_plus_connect_timeout}, + ) + connection = engine.raw_connection() + try: + connection.driver_connection.timeout = settings.t_plus_query_timeout + report = _inspect_connection(connection, args.expected_version) + finally: + connection.close() + _inspect_adapter(engine, report) + except Exception as exc: + # 驱动异常可能包含完整连接串，报告仅保留类型。 + report = {"compatible": False, "error_type": type(exc).__name__} + finally: + if engine is not None: + engine.dispose() args.report.parent.mkdir(parents=True, exist_ok=True) args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(report, ensure_ascii=False))