test_service_password.py #1

#!/usr/bin/env python3
"""Service account initialization keeps reset policy and checks both mutations."""
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest

ROOT = Path(__file__).resolve().parents[1]
HELPER = ROOT / "Server/Unix/p4/common/bin/sdp_service_user.sh"
BASH = os.environ.get("SDP_TEST_BASH", shutil.which("bash"))


class ServicePassword(unittest.TestCase):
    def run_case(self, policy="1", failure=""):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            stub = root / "p4"
            stub.write_text("#!" + sys.executable + r'''
import json,os,sys
from pathlib import Path
args=sys.argv[1:]
if args[:2] == ['-ztag','configure']:
    if os.environ['FAILURE']=='query': sys.exit(9)
    print('... Value '+os.environ['POLICY'])
else:
    data=sys.stdin.read()
    with Path(os.environ['TRACE']).open('a') as out: out.write(json.dumps({'argv':args,'input':data})+'\n')
    phase='self' if args[:1]==['-u'] else 'initial'
    if os.environ['FAILURE']==phase: sys.exit(7)
    print('Password updated.')
''')
            stub.chmod(0o755)
            env = dict(os.environ, P4BIN=str(stub), TRACE=str(root / "trace"), POLICY=policy, FAILURE=failure)
            result = subprocess.run([BASH, "-c", 'source "$1"; sdp_initialize_service_password svc_fixture "$2"', "test", str(HELPER), "Final Password * 2026!"], env=env, capture_output=True, text=True)
            trace = [json.loads(x) for x in (root / "trace").read_text().splitlines()] if (root / "trace").exists() else []
            self.assertNotIn("Final Password", result.stdout + result.stderr)
            return result, trace

    def test_non_expiring_policy_preserves_direct_initialization(self):
        result, calls = self.run_case("0")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(calls, [{"argv": ["passwd", "svc_fixture"], "input": "Final Password * 2026!\n" * 2}])

    def test_expiring_policy_completes_own_reset_without_password_arguments(self):
        result, calls = self.run_case()
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual([r["argv"] for r in calls], [["passwd", "svc_fixture"], ["-u", "svc_fixture", "passwd"]])
        initial = calls[0]["input"].splitlines()
        self.assertEqual(initial[0], initial[1])
        self.assertRegex(initial[0], r"^SdpInit[0-9a-f]{48}!$")
        self.assertEqual(calls[1]["input"].splitlines(), [initial[0], "Final Password * 2026!", "Final Password * 2026!"])

    def test_query_failure_changes_no_password(self):
        result, calls = self.run_case(failure="query")
        self.assertEqual(result.returncode, 9)
        self.assertEqual(calls, [])

    def test_invalid_policy_changes_no_password(self):
        result, calls = self.run_case("unknown")
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual(calls, [])

    def test_initial_failure_prevents_self_change(self):
        result, calls = self.run_case(failure="initial")
        self.assertEqual(result.returncode, 7)
        self.assertEqual(len(calls), 1)

    def test_self_change_failure_is_not_success(self):
        result, calls = self.run_case(failure="self")
        self.assertEqual(result.returncode, 7)
        self.assertEqual(len(calls), 2)

    def test_installer_service_password_path_uses_checked_helper(self):
        # Execute the actual branch with an encrypted input and a failed helper;
        # this catches an unchecked password failure in configure_new_server.
        source = (ROOT / "Server/Unix/setup/configure_new_server.sh").read_text()
        start = source.index('      source "$P4CBIN/sdp_service_user.sh"')
        end = source.index('      unset ServicePassword', start) + len('      unset ServicePassword')
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            (root / "sdp_service_user.sh").write_text('sdp_initialize_service_password() { return 7; }\n')
            password = root / "password"
            password.write_text("installer fixture password")
            env = dict(os.environ, P4CBIN=str(root), P4SERVICEUSER="svc_fixture", CleartextPasswordFile=str(password), EncryptedPasswordFile=str(root / "absent"))
            result = subprocess.run([BASH, "-c", 'ErrorCount=0; errmsg(){ ErrorCount=$((ErrorCount+1)); }; bail(){ exit 99; };\n' + source[start:end] + '\nexit "$ErrorCount"'], env=env, capture_output=True, text=True)
            self.assertEqual(result.returncode, 1)
            self.assertNotIn("installer fixture password", result.stdout + result.stderr)


if __name__ == "__main__":
    unittest.main()