redmine/reset_helpdesk_mail_settings.py

#!/usr/bin/env python3
"""Reset RedmineUP Helpdesk mail settings on the LAN test Redmine instance.

This is intended to be run after importing a production database into the test
instance. It rewrites every active project's incoming/outgoing Helpdesk mail
settings so test mail flows through Mailpit and imported real credentials cannot
be used accidentally.
"""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any


DEFAULT_SSH_HOST = "reddev@192.168.50.170"
DEFAULT_SSH_KEY = Path("/tmp/reddev")
DEFAULT_REMOTE_REDMINE = "/usr/share/redmine"
DEFAULT_MAILPIT_HOST = "192.168.1.105"

MAIL_SETTING_NAMES = [
    "helpdesk_protocol",
    "helpdesk_host",
    "helpdesk_port",
    "helpdesk_username",
    "helpdesk_password",
    "helpdesk_use_ssl",
    "helpdesk_imap_folder",
    "helpdesk_move_on_success",
    "helpdesk_move_on_failure",
    "helpdesk_apop",
    "helpdesk_delete_unprocessed",
    "helpdesk_smtp_use_default_settings",
    "helpdesk_smtp_server",
    "helpdesk_smtp_port",
    "helpdesk_smtp_domain",
    "helpdesk_smtp_authentication",
    "helpdesk_smtp_username",
    "helpdesk_smtp_password",
    "helpdesk_smtp_ssl",
    "helpdesk_smtp_tls",
    "helpdesk_answer_from",
]

SECRET_NAMES = {"helpdesk_password", "helpdesk_smtp_password"}


class ResetError(RuntimeError):
    pass


@dataclass(frozen=True)
class RemoteRedmine:
    ssh_host: str
    ssh_key: Path
    remote_redmine: str

    def mysql_json_lines(self, sql: str) -> list[dict[str, Any]]:
        stdout = self.mysql(sql)
        rows: list[dict[str, Any]] = []
        for line in stdout.splitlines():
            if not line.strip():
                continue
            try:
                rows.append(json.loads(bytes.fromhex(line.strip()).decode("utf-8")))
            except (ValueError, json.JSONDecodeError) as exc:
                raise ResetError(f"Remote query returned an unexpected row: {line[:200]}") from exc
        return rows

    def mysql(self, sql: str) -> str:
        command = [
            "ssh",
            "-i",
            str(self.ssh_key),
            "-o",
            "IdentitiesOnly=yes",
            self.ssh_host,
            self._mysql_runner_command(),
        ]
        try:
            result = subprocess.run(
                command,
                input=sql,
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                check=False,
            )
        except OSError as exc:
            raise ResetError(f"Could not run ssh: {exc}") from exc

        if result.returncode != 0:
            raise ResetError(result.stderr.strip() or "Remote MySQL command failed.")
        return result.stdout

    def _mysql_runner_command(self) -> str:
        ruby = (
            "require 'yaml'; "
            "c = YAML.load_file('config/database.yml')['production']; "
            "ENV['MYSQL_PWD'] = c['password'].to_s; "
            "args = ['--batch', '--raw', '--quick', '--skip-column-names', "
            "'--default-character-set=utf8', '-h', c['host'].to_s, "
            "'-P', (c['port'] || 3306).to_s, '-u', c['username'].to_s, c['database'].to_s]; "
            "exec('mysql', *args)"
        )
        return f"cd {shell_quote(self.remote_redmine)} && ruby -e {shell_quote(ruby)}"


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Reset Helpdesk mail settings for all active projects."
    )
    parser.add_argument("--ssh-host", default=os.getenv("REDMINE_SSH_HOST", DEFAULT_SSH_HOST))
    parser.add_argument("--ssh-key", type=Path, default=Path(os.getenv("REDMINE_SSH_KEY", str(DEFAULT_SSH_KEY))))
    parser.add_argument("--remote-redmine", default=os.getenv("REDMINE_REMOTE_PATH", DEFAULT_REMOTE_REDMINE))
    parser.add_argument("--mailpit-host", default=DEFAULT_MAILPIT_HOST, help="Host Redmine should use to reach Mailpit.")
    parser.add_argument("--pop3-port", type=int, default=1110)
    parser.add_argument("--smtp-port", type=int, default=1025)
    parser.add_argument("--username", default="test")
    parser.add_argument("--password", default="testpass")
    parser.add_argument("--smtp-domain", default="example.test")
    parser.add_argument(
        "--from-pattern",
        default="helpdesk-{identifier}@example.test",
        help="Pattern for helpdesk_answer_from. Available fields: {id}, {identifier}, {name}.",
    )
    parser.add_argument(
        "--project",
        action="append",
        default=[],
        help="Optional project id or identifier to limit changes. Can be passed more than once.",
    )
    parser.add_argument("--dry-run", action="store_true", help="Show affected projects and settings without writing.")
    args = parser.parse_args()

    remote = RemoteRedmine(args.ssh_host, args.ssh_key, args.remote_redmine)

    try:
        projects = find_active_projects(remote, args.project)
        if not projects:
            print("No active projects matched the requested filters.")
            return 0

        print(f"Matched {len(projects)} active project(s):")
        for project in projects:
            print(f"  - #{project['id']} {project['identifier']} ({project['name']})")

        values = build_values(args, projects)
        if args.dry_run:
            print("\nDry run. Planned settings:")
            print_plan(values)
            return 0

        apply_values(remote, values)
        print(f"\nUpdated {len(values)} setting row(s) across {len(projects)} project(s).")
        print("Password values were written but not displayed.")
        return 0
    except ResetError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1


def find_active_projects(remote: RemoteRedmine, filters: list[str]) -> list[dict[str, Any]]:
    where = ["p.status = 1"]
    if filters:
        clauses = []
        for value in filters:
            if str(value).isdigit():
                clauses.append(f"p.id = {sql_int(value)}")
            clauses.append(f"p.identifier = {sql_string(value)}")
        where.append("(" + " OR ".join(clauses) + ")")

    return remote.mysql_json_lines(
        f"""
SELECT HEX(CAST(JSON_OBJECT(
  'id', p.id,
  'identifier', p.identifier,
  'name', p.name
) AS CHAR)) AS document
FROM projects p
WHERE {' AND '.join(where)}
ORDER BY p.identifier;
"""
    )


def build_values(args: argparse.Namespace, projects: list[dict[str, Any]]) -> list[tuple[int, str, str]]:
    rows: list[tuple[int, str, str]] = []
    for project in projects:
        project_id = int(project["id"])
        answer_from = args.from_pattern.format(
            id=project_id,
            identifier=project["identifier"],
            name=project["name"],
        )
        settings = {
            "helpdesk_protocol": "pop3",
            "helpdesk_host": args.mailpit_host,
            "helpdesk_port": str(args.pop3_port),
            "helpdesk_username": args.username,
            "helpdesk_password": args.password,
            "helpdesk_use_ssl": "0",
            "helpdesk_imap_folder": "",
            "helpdesk_move_on_success": "",
            "helpdesk_move_on_failure": "",
            "helpdesk_apop": "0",
            "helpdesk_delete_unprocessed": "0",
            # RedmineUP's UI label is confusing: 1 means use the custom SMTP block.
            "helpdesk_smtp_use_default_settings": "1",
            "helpdesk_smtp_server": args.mailpit_host,
            "helpdesk_smtp_port": str(args.smtp_port),
            "helpdesk_smtp_domain": args.smtp_domain,
            "helpdesk_smtp_authentication": "",
            "helpdesk_smtp_username": "",
            "helpdesk_smtp_password": "",
            "helpdesk_smtp_ssl": "0",
            "helpdesk_smtp_tls": "0",
            "helpdesk_answer_from": answer_from,
        }
        rows.extend((project_id, name, settings[name]) for name in MAIL_SETTING_NAMES)
    return rows


def apply_values(remote: RemoteRedmine, rows: list[tuple[int, str, str]]) -> None:
    statements = ["START TRANSACTION;"]
    for project_id, name, value in rows:
        project_id_sql = sql_int(project_id)
        name_sql = sql_string(name)
        value_sql = sql_string(value)
        statements.append(
            f"""
UPDATE contacts_settings
SET value = {value_sql}, updated_on = UTC_TIMESTAMP()
WHERE project_id = {project_id_sql}
  AND name = {name_sql};
INSERT INTO contacts_settings (project_id, name, value, updated_on)
SELECT {project_id_sql}, {name_sql}, {value_sql}, UTC_TIMESTAMP()
WHERE NOT EXISTS (
  SELECT 1
  FROM contacts_settings
  WHERE project_id = {project_id_sql}
    AND name = {name_sql}
);
"""
        )
    statements.append("COMMIT;")

    remote.mysql("\n".join(statements))


def print_plan(rows: list[tuple[int, str, str]]) -> None:
    current_project_id: int | None = None
    for project_id, name, value in rows:
        if project_id != current_project_id:
            current_project_id = project_id
            print(f"\nProject #{project_id}")
        display_value = "<redacted>" if name in SECRET_NAMES else value
        print(f"  {name} = {display_value}")


def sql_int(value: Any) -> int:
    try:
        return max(0, int(value))
    except (TypeError, ValueError):
        return 0


def sql_string(value: Any) -> str:
    return "'" + str(value).replace("\\", "\\\\").replace("'", "\\'") + "'"


def shell_quote(value: str) -> str:
    return "'" + value.replace("'", "'\"'\"'") + "'"


if __name__ == "__main__":
    raise SystemExit(main())