From 23aa2fb1921dab5a43b314ecf621971abf2560a3 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 3 Jul 2026 17:53:07 -0400 Subject: [PATCH] fix: harden profiles migration helper --- migrate_profiles.py | 116 +++++++++++++++----- tests/test_migrate_profiles.py | 186 +++++++++++++++++++++++++-------- 2 files changed, 232 insertions(+), 70 deletions(-) diff --git a/migrate_profiles.py b/migrate_profiles.py index 42e8292..3613c04 100755 --- a/migrate_profiles.py +++ b/migrate_profiles.py @@ -20,6 +20,43 @@ if PROJECT_ROOT not in sys.path: import gitea_config +AUTHOR_DEFAULT_ALLOWED = ["read", "branch", "commit", "push", "open_pr", "comment"] +AUTHOR_DEFAULT_FORBIDDEN = ["approve", "request_changes", "merge"] +REVIEWER_DEFAULT_ALLOWED = [ + "read", "review", "comment", "approve", "request_changes", "merge" +] +REVIEWER_DEFAULT_FORBIDDEN = ["branch", "commit", "push", "open_pr"] + + +def infer_role(name, execution_profile): + """Return the unambiguous role for a legacy profile name, or None.""" + haystack = f"{name} {execution_profile or ''}".lower() + has_author = "author" in haystack + has_reviewer = "reviewer" in haystack + if has_author == has_reviewer: + return None + return "reviewer" if has_reviewer else "author" + + +def migration_summary(v2_data): + """Return a redacted summary of the migrated config.""" + environments = v2_data.get("environments", {}) + service_count = 0 + identity_count = 0 + for env in environments.values(): + services = env.get("services", {}) + service_count += len(services) + for service in services.values(): + identity_count += len(service.get("identities", {})) + return { + "version": v2_data.get("version"), + "environments": len(environments), + "services": service_count, + "identities": identity_count, + "aliases": len(v2_data.get("aliases", {})), + } + + def migrate_v1_to_v2(v1_data): """Convert version 1 profiles.json format to version 2 environments format.""" environments = {} @@ -42,43 +79,62 @@ def migrate_v1_to_v2(v1_data): env_name = name ident_name = "author" - # Determine role and identity based on name / execution_profile - role = "author" + # Determine role and identity based on name / execution_profile. + # Ambiguous profiles may still migrate only when they carry explicit + # permissions; otherwise role-based defaults could widen permissions. exec_prof = prof.get("execution_profile") or "" - if "reviewer" in name or "reviewer" in exec_prof: - role = "reviewer" + role = infer_role(name, exec_prof) + if role == "reviewer": ident_name = "reviewer" - elif "author" in name or "author" in exec_prof: - role = "author" + elif role == "author": ident_name = "author" + else: + role = prof.get("role") + if role not in (None, "author", "reviewer"): + raise ValueError( + f"Profile '{name}' has unsupported role {role!r}" + ) # Construct identity block identity_data = { - "role": role, "username": prof.get("username"), "auth": prof.get("auth"), } + if role: + identity_data["role"] = role if prof.get("execution_profile"): identity_data["execution_profile"] = prof["execution_profile"] - + # Set audit label (default to old name to preserve context) identity_data["audit_label"] = prof.get("audit_label") or name - # Populate capabilities based on role - if role == "author": - identity_data["allowed_operations"] = [ - "read", "branch", "commit", "push", "open_pr", "comment" - ] - identity_data["forbidden_operations"] = [ - "approve", "request_changes", "merge" - ] + has_allowed = "allowed_operations" in prof + has_forbidden = "forbidden_operations" in prof + if has_allowed != has_forbidden: + raise ValueError( + f"Profile '{name}' must define both allowed_operations and " + "forbidden_operations, or neither (fail closed)" + ) + if has_allowed: + allowed = prof.get("allowed_operations") + forbidden = prof.get("forbidden_operations") + if not isinstance(allowed, list) or not isinstance(forbidden, list): + raise ValueError( + f"Profile '{name}' operation fields must be lists" + ) + identity_data["allowed_operations"] = list(allowed) + identity_data["forbidden_operations"] = list(forbidden) + elif role == "author": + identity_data["allowed_operations"] = list(AUTHOR_DEFAULT_ALLOWED) + identity_data["forbidden_operations"] = list(AUTHOR_DEFAULT_FORBIDDEN) + elif role == "reviewer": + identity_data["allowed_operations"] = list(REVIEWER_DEFAULT_ALLOWED) + identity_data["forbidden_operations"] = list(REVIEWER_DEFAULT_FORBIDDEN) else: - identity_data["allowed_operations"] = [ - "read", "review", "comment", "approve", "request_changes", "merge" - ] - identity_data["forbidden_operations"] = [ - "branch", "commit", "push", "open_pr" - ] + raise ValueError( + f"Profile '{name}' has no explicit operation lists and no " + "unambiguous author/reviewer role marker (fail closed)" + ) # Nest inside environments/services structure env = environments.setdefault(env_name, {}) @@ -100,7 +156,7 @@ def migrate_v1_to_v2(v1_data): alias_target = f"{env_name}.gitea.{ident_name}" if name != alias_target: aliases[name] = alias_target - + # Extra convenience alias for standard old-profile compatibility (e.g. prgs-author) convenience_alias = f"{env_name}-{ident_name}" if convenience_alias != alias_target and convenience_alias not in aliases: @@ -191,11 +247,15 @@ def main(): if not args.write: print("=== DRY-RUN MODE (No files modified) ===") - print(f"Would read from: {input_path}") - print(f"Would create backup at: {backup_path}") - print(f"Would write v2 config to: {output_path}") - print("\nGenerated v2 config:") - print(json.dumps(v2_data, indent=2)) + print("Generated v2 config validated successfully.") + print("Only aggregate counts are shown.") + summary = migration_summary(v2_data) + print("Summary:") + print(f" version: {summary['version']}") + print(f" environments: {summary['environments']}") + print(f" services: {summary['services']}") + print(f" identities: {summary['identities']}") + print(f" aliases: {summary['aliases']}") sys.exit(0) # Write Mode: Create Backup first diff --git a/tests/test_migrate_profiles.py b/tests/test_migrate_profiles.py index b6725ad..84bcaa9 100644 --- a/tests/test_migrate_profiles.py +++ b/tests/test_migrate_profiles.py @@ -25,34 +25,43 @@ class TestMigrateProfiles(unittest.TestCase): "version": 1, "profiles": { "prgs": { - "base_url": "https://gitea.prgs.cc", - "username": "jcwalker3", - "auth": { - "type": "keychain", - "id": "prgs-gitea-token" - }, - "default_owner": "Scaled-Tech-Consulting", - "execution_profile": "personal-prgs" + "base_url": "redacted-prgs-service", + "username": "jcwalker3", + "auth": { + "type": "keychain", + "id": "redacted-author-ref" + }, + "default_owner": "Scaled-Tech-Consulting", + "execution_profile": "personal-prgs", + "allowed_operations": ["read", "comment"], + "forbidden_operations": ["approve", "merge"] }, "mdcps": { - "base_url": "https://gitea.dadeschools.net", - "username": "913443", - "auth": { - "type": "keychain", - "id": "mdcps-gitea-token" - }, - "default_owner": "Contractor", - "execution_profile": "mdcps" + "base_url": "redacted-mdcps-service", + "username": "913443", + "auth": { + "type": "keychain", + "id": "redacted-mdcps-ref" + }, + "default_owner": "Contractor", + "execution_profile": "mdcps", + "allowed_operations": ["read"], + "forbidden_operations": ["merge"] }, "prgs-reviewer": { - "base_url": "https://gitea.prgs.cc", - "username": "sysadmin", - "auth": { - "type": "keychain", - "id": "prgs-gitea-reviewer-token" - }, - "default_owner": "Scaled-Tech-Consulting", - "execution_profile": "prgs-reviewer" + "base_url": "redacted-prgs-service", + "username": "sysadmin", + "auth": { + "type": "keychain", + "id": "redacted-reviewer-ref" + }, + "default_owner": "Scaled-Tech-Consulting", + "execution_profile": "prgs-reviewer", + "allowed_operations": [ + "read", "review", "comment", "approve", + "request_changes", "merge" + ], + "forbidden_operations": ["branch", "commit", "push", "open_pr"] } } } @@ -67,34 +76,34 @@ class TestMigrateProfiles(unittest.TestCase): """Test the structural transformation and capability mapping.""" v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content) self.assertEqual(v2_data["version"], 2) - + # Check environment structure envs = v2_data["environments"] self.assertIn("prgs", envs) self.assertIn("mdcps", envs) - + # Check service and identity structure prgs_gitea = envs["prgs"]["services"]["gitea"] - self.assertEqual(prgs_gitea["base_url"], "https://gitea.prgs.cc") + self.assertEqual(prgs_gitea["base_url"], "redacted-prgs-service") self.assertEqual(prgs_gitea["default_owner"], "Scaled-Tech-Consulting") - + author = prgs_gitea["identities"]["author"] - self.assertEqual(author["role"], "author") self.assertEqual(author["username"], "jcwalker3") - self.assertEqual(author["auth"]["id"], "prgs-gitea-token") - self.assertIn("push", author["allowed_operations"]) - + self.assertEqual(author["auth"]["id"], "redacted-author-ref") + self.assertEqual(author["allowed_operations"], ["read", "comment"]) + self.assertEqual(author["forbidden_operations"], ["approve", "merge"]) + reviewer = prgs_gitea["identities"]["reviewer"] self.assertEqual(reviewer["role"], "reviewer") self.assertEqual(reviewer["username"], "sysadmin") - self.assertEqual(reviewer["auth"]["id"], "prgs-gitea-reviewer-token") + self.assertEqual(reviewer["auth"]["id"], "redacted-reviewer-ref") self.assertIn("merge", reviewer["allowed_operations"]) def test_alias_generation(self): """Test that aliases are correctly generated to support old profile names.""" v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content) aliases = v2_data["aliases"] - + self.assertEqual(aliases["prgs"], "prgs.gitea.author") self.assertEqual(aliases["prgs-author"], "prgs.gitea.author") self.assertEqual(aliases["prgs-reviewer"], "prgs.gitea.reviewer") @@ -131,20 +140,113 @@ class TestMigrateProfiles(unittest.TestCase): with self.assertRaises(SystemExit) as cm: migrate_profiles.main() self.assertEqual(cm.exception.code, 0) - + self.assertFalse(os.path.exists(output_file)) self.assertFalse(os.path.exists(f"{self.input_file}.bak")) - + stdout_output = mock_stdout.getvalue() self.assertIn("DRY-RUN MODE", stdout_output) self.assertIn("version", stdout_output) - self.assertIn("environments", stdout_output) + self.assertIn("identities", stdout_output) + self.assertIn("aliases", stdout_output) + self.assertNotIn("redacted-prgs-service", stdout_output) + self.assertNotIn("redacted-mdcps-service", stdout_output) + self.assertNotIn("redacted-author-ref", stdout_output) + self.assertNotIn("redacted-mdcps-ref", stdout_output) + self.assertNotIn("redacted-reviewer-ref", stdout_output) + self.assertNotIn("keychain", stdout_output) + self.assertNotIn("auth", stdout_output) + + def test_dry_run_hides_token_like_values(self): + """Verify dry-run summary does not expose token-like auth metadata.""" + sensitive = json.loads(json.dumps(self.v1_content)) + sensitive["profiles"]["prgs"]["auth"]["id"] = "super-secret-token-value" + with open(self.input_file, "w") as f: + json.dump(sensitive, f) + + test_args = ["migrate_profiles.py", "-i", self.input_file] + with patch("sys.stdout", new_callable=StringIO) as mock_stdout: + with patch.object(sys, "argv", test_args): + with self.assertRaises(SystemExit) as cm: + migrate_profiles.main() + self.assertEqual(cm.exception.code, 0) + + stdout_output = mock_stdout.getvalue() + self.assertNotIn("super-secret-token-value", stdout_output) + self.assertNotIn("token", stdout_output.lower()) + + def test_explicit_operations_are_preserved(self): + """Explicit v1 permissions must not be replaced by role defaults.""" + v1_data = json.loads(json.dumps(self.v1_content)) + v1_data["profiles"]["prgs-reviewer"]["allowed_operations"] = ["read"] + v1_data["profiles"]["prgs-reviewer"]["forbidden_operations"] = ["merge"] + + v2_data = migrate_profiles.migrate_v1_to_v2(v1_data) + reviewer = ( + v2_data["environments"]["prgs"]["services"]["gitea"] + ["identities"]["reviewer"] + ) + self.assertEqual(reviewer["allowed_operations"], ["read"]) + self.assertEqual(reviewer["forbidden_operations"], ["merge"]) + + def test_inferred_role_defaults_only_when_unambiguous(self): + """Role defaults are allowed only for clear author/reviewer profiles.""" + v1_data = { + "version": 1, + "profiles": { + "prgs-author": { + "base_url": "redacted-prgs-service", + "username": "jcwalker3", + "auth": {"type": "keychain", "id": "hidden-author-ref"}, + "execution_profile": "prgs-author", + } + } + } + + v2_data = migrate_profiles.migrate_v1_to_v2(v1_data) + author = ( + v2_data["environments"]["prgs"]["services"]["gitea"] + ["identities"]["author"] + ) + self.assertEqual( + author["allowed_operations"], + migrate_profiles.AUTHOR_DEFAULT_ALLOWED, + ) + self.assertEqual( + author["forbidden_operations"], + migrate_profiles.AUTHOR_DEFAULT_FORBIDDEN, + ) + + def test_ambiguous_permission_source_fails_closed(self): + """A profile without explicit permissions or clear role must not widen.""" + v1_data = { + "version": 1, + "profiles": { + "prgs": { + "base_url": "redacted-prgs-service", + "username": "jcwalker3", + "auth": {"type": "keychain", "id": "hidden-author-ref"}, + "execution_profile": "personal-prgs", + } + } + } + + with self.assertRaisesRegex(ValueError, "fail closed"): + migrate_profiles.migrate_v1_to_v2(v1_data) + + def test_partial_permission_source_fails_closed(self): + """Allowed without forbidden, or vice versa, is ambiguous.""" + v1_data = json.loads(json.dumps(self.v1_content)) + del v1_data["profiles"]["prgs"]["forbidden_operations"] + + with self.assertRaisesRegex(ValueError, "fail closed"): + migrate_profiles.migrate_v1_to_v2(v1_data) def test_write_mode_and_backup(self): """Verify that write mode creates a backup and correctly saves the validated config.""" output_file = os.path.join(self.temp_dir, "migrated.json") backup_file = os.path.join(self.temp_dir, "profiles_backup.json.bak") - + test_args = [ "migrate_profiles.py", "-i", self.input_file, @@ -154,14 +256,14 @@ class TestMigrateProfiles(unittest.TestCase): ] with patch.object(sys, "argv", test_args): migrate_profiles.main() - + # Verify backup exists and matches original v1 config self.assertTrue(os.path.exists(backup_file)) with open(backup_file, "r") as f: backup_data = json.load(f) self.assertEqual(backup_data["version"], 1) self.assertIn("prgs", backup_data["profiles"]) - + # Verify migrated v2 config exists and validates self.assertTrue(os.path.exists(output_file)) with open(output_file, "r") as f: @@ -175,7 +277,7 @@ class TestMigrateProfiles(unittest.TestCase): bad_json_file = os.path.join(self.temp_dir, "bad.json") with open(bad_json_file, "w") as f: f.write("{invalid-json}") - + test_args = ["migrate_profiles.py", "-i", bad_json_file] with patch.object(sys, "argv", test_args): with self.assertRaises(SystemExit) as cm: @@ -185,7 +287,7 @@ class TestMigrateProfiles(unittest.TestCase): bad_version_file = os.path.join(self.temp_dir, "bad_version.json") with open(bad_version_file, "w") as f: json.dump({"version": 3, "profiles": {}}, f) - + test_args = ["migrate_profiles.py", "-i", bad_version_file] with patch.object(sys, "argv", test_args): with self.assertRaises(SystemExit) as cm: