Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 30 additions & 11 deletions src/scripts/meza.py
Original file line number Diff line number Diff line change
Expand Up @@ -366,22 +366,41 @@ def request_lock_for_deploy(env):
f.write(f"{pid}\n{timestamp}")
f.close()

# Improved group detection and fallback for lock file permissions
lock_owner, lock_group = get_deploy_lock_owner_and_group()
meza_chown(lock_file, lock_owner, lock_group)
os.chmod(lock_file, 0o664)

return {"pid": pid, "timestamp": timestamp}


def get_deploy_lock_owner_and_group():
"""
Resolve a valid owner and group for deploy lock files.

Returns:
tuple: Username and group name for lock file ownership.
"""
lock_owner = 'meza-ansible'
try:
grp.getgrnam('apache')
meza_chown(lock_file, 'meza-ansible', 'apache')
os.chmod(lock_file, 0o664)
pwd.getpwnam(lock_owner)
except KeyError:
lock_owner = pwd.getpwuid(os.getuid()).pw_name
print(f'User "meza-ansible" not found. Using "{lock_owner}" as fallback.')

preferred_groups = ['apache', 'www-data', 'wheel']

for group_name in preferred_groups:
try:
grp.getgrnam('www-data') # Debian/Ubuntu fallback
meza_chown(lock_file, 'meza-ansible', 'www-data')
os.chmod(lock_file, 0o664)
grp.getgrnam(group_name)
if group_name == 'wheel':
print('Neither apache nor www-data group exists. Using "wheel" as fallback.')
return lock_owner, group_name
except KeyError:
print('Neither apache nor www-data group exists. Using "wheel" as fallback.')
meza_chown(lock_file, 'meza-ansible', 'wheel')
os.chmod(lock_file, 0o664)
continue

return {"pid": pid, "timestamp": timestamp}
lock_group = grp.getgrgid(os.getgid()).gr_name
print(f'Neither apache, www-data, nor wheel group exists. Using "{lock_group}" as fallback.')
return lock_owner, lock_group


def unlock_deploy(env):
Expand Down
78 changes: 78 additions & 0 deletions tests/unit/test_deploy_lock_fallback.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
#!/usr/bin/env python3
"""Regression tests for deploy lock ownership fallback logic."""

import importlib.util
import pathlib
import shutil
import tempfile
import types
import unittest
from unittest import mock


class DeployLockFallbackTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.repo_root = pathlib.Path(__file__).resolve().parents[2]
cls.temp_layout = tempfile.TemporaryDirectory()
module_root = pathlib.Path(cls.temp_layout.name) / "opt" / "meza"
scripts_dir = module_root / "src" / "scripts"
i18n_dir = module_root / "config" / "i18n"

scripts_dir.mkdir(parents=True, exist_ok=True)
i18n_dir.mkdir(parents=True, exist_ok=True)

shutil.copy2(cls.repo_root / "src" / "scripts" / "meza.py", scripts_dir / "meza.py")
shutil.copy2(cls.repo_root / "config" / "i18n" / "en.yml", i18n_dir / "en.yml")

module_path = scripts_dir / "meza.py"
spec = importlib.util.spec_from_file_location(
"meza_module_under_test",
module_path,
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
cls.meza = module

@classmethod
def tearDownClass(cls):
cls.temp_layout.cleanup()

def test_request_lock_uses_fallback_user_and_group(self):
with tempfile.TemporaryDirectory() as meza_data_dir:
self.meza.defaults["m_meza_data"] = meza_data_dir

def fake_getpwnam(username):
if username == "meza-ansible":
raise KeyError(username)
raise AssertionError(f"Unexpected user lookup: {username}")

def fake_getgrnam(groupname):
if groupname in ("apache", "www-data"):
raise KeyError(groupname)
if groupname == "wheel":
return types.SimpleNamespace(gr_gid=10)
raise AssertionError(f"Unexpected group lookup: {groupname}")

with mock.patch.object(self.meza.os, "getpid", return_value=1234), \
mock.patch.object(self.meza.pwd, "getpwnam", side_effect=fake_getpwnam), \
mock.patch.object(
self.meza.pwd,
"getpwuid",
return_value=types.SimpleNamespace(pw_name="root"),
), \
mock.patch.object(self.meza.grp, "getgrnam", side_effect=fake_getgrnam), \
mock.patch.object(self.meza, "meza_chown") as mocked_chown, \
mock.patch.object(self.meza.os, "chmod") as mocked_chmod:
result = self.meza.request_lock_for_deploy("monolith")

lock_file = pathlib.Path(meza_data_dir) / "env-monolith-deploy.lock"

self.assertTrue(lock_file.exists())
self.assertEqual("1234", result["pid"])
mocked_chown.assert_called_once_with(str(lock_file), "root", "wheel")
mocked_chmod.assert_called_once_with(str(lock_file), 0o664)


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