diff --git a/crates/unica-coder/src/application/mod.rs b/crates/unica-coder/src/application/mod.rs index b8503a9b..7dff731a 100644 --- a/crates/unica-coder/src/application/mod.rs +++ b/crates/unica-coder/src/application/mod.rs @@ -10061,6 +10061,7 @@ mod tests { "cccccccc-cccc-cccc-cccc-cccccccccccc", ), ); + write_support_test_vendor_payload(&workspace); let before = std::fs::read_to_string(&bin_path).unwrap(); let mut args = Map::new(); args.insert( @@ -10081,6 +10082,100 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[test] + fn support_edit_missing_vendor_payload_blocks_preview_and_apply_before_side_effects() { + let bin = support_test_parent_configurations_bin( + "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", + "cccccccc-cccc-cccc-cccc-cccccccccccc", + ) + .replace("{6,0,", "{6,1,"); + let (root, workspace, bin_path) = + support_test_workspace("unica-support-edit-missing-vendor-payload", bin); + let before = std::fs::read(&bin_path).unwrap(); + let state_path = workspace.join(".build/unica/state.json"); + let mut args = Map::from_iter([ + ( + "cwd".to_string(), + Value::String(workspace.display().to_string()), + ), + ("Path".to_string(), Value::String("src".to_string())), + ("Capability".to_string(), Value::String("on".to_string())), + ]); + let mut results = Vec::new(); + + for dry_run in [false, true] { + args.insert("dryRun".to_string(), Value::Bool(dry_run)); + let result = UnicaApplication::new() + .call_tool("unica.support.edit", &args) + .unwrap(); + + assert!(!result.ok, "dryRun={dry_run}: {result:?}"); + assert!( + result.errors.join("\n").contains("VendorConf.cf"), + "dryRun={dry_run}: {result:?}" + ); + assert!(result.cache.events.is_empty(), "{result:?}"); + assert_eq!(std::fs::read(&bin_path).unwrap(), before); + assert!(!state_path.exists(), "dryRun={dry_run}"); + results.push(result); + } + assert_support_guard_block_parity(&results[0], &results[1]); + + std::fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn support_edit_unreadable_vendor_payload_blocks_preview_and_apply_before_side_effects() { + let bin = support_test_parent_configurations_bin( + "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", + "cccccccc-cccc-cccc-cccc-cccccccccccc", + ) + .replace("{6,0,", "{6,1,"); + let (root, workspace, bin_path) = + support_test_workspace("unica-support-edit-unreadable-vendor-payload", bin); + write_support_test_vendor_payload(&workspace); + let vendor_payload = workspace.join("src/Ext/ParentConfigurations/VendorConf.cf"); + let before = std::fs::read(&bin_path).unwrap(); + let state_path = workspace.join(".build/unica/state.json"); + if !set_unix_mode_for_test(&vendor_payload, 0o000).unwrap() { + eprintln!("[SKIPPED FIXTURE] Unix permission modes are unsupported on this host"); + std::fs::remove_dir_all(root).unwrap(); + return; + } + let mut args = Map::from_iter([ + ( + "cwd".to_string(), + Value::String(workspace.display().to_string()), + ), + ("Path".to_string(), Value::String("src".to_string())), + ("Capability".to_string(), Value::String("on".to_string())), + ]); + let mut results = Vec::new(); + + for dry_run in [false, true] { + args.insert("dryRun".to_string(), Value::Bool(dry_run)); + let result = UnicaApplication::new() + .call_tool("unica.support.edit", &args) + .unwrap(); + + assert!(!result.ok, "dryRun={dry_run}: {result:?}"); + assert!( + result.errors.join("\n").contains("VendorConf.cf"), + "dryRun={dry_run}: {result:?}" + ); + assert!(result.cache.events.is_empty(), "{result:?}"); + assert_eq!(std::fs::read(&bin_path).unwrap(), before); + assert!(!state_path.exists(), "dryRun={dry_run}"); + results.push(result); + } + assert_support_guard_block_parity(&results[0], &results[1]); + + assert!(set_unix_mode_for_test(&vendor_payload, 0o600).unwrap()); + std::fs::remove_dir_all(root).unwrap(); + } + #[test] fn support_edit_capability_on_enables_global_editing() { let bin = support_test_parent_configurations_bin( @@ -10091,6 +10186,7 @@ mod tests { .replace("{6,0,", "{6,1,"); let (root, workspace, _bin_path) = support_test_workspace("unica-support-edit-capability-on", bin); + write_support_test_vendor_payload(&workspace); let mut args = Map::new(); args.insert( "cwd".to_string(), @@ -10130,6 +10226,7 @@ mod tests { "cccccccc-cccc-cccc-cccc-cccccccccccc", ), ); + write_support_test_vendor_payload(&workspace); let mut args = Map::new(); args.insert( "cwd".to_string(), @@ -11681,6 +11778,12 @@ mod tests { .unwrap(); } + fn write_support_test_vendor_payload(workspace: &std::path::Path) { + let vendor_dir = workspace.join("src/Ext/ParentConfigurations"); + std::fs::create_dir_all(&vendor_dir).unwrap(); + std::fs::write(vendor_dir.join("VendorConf.cf"), b"platform vendor payload").unwrap(); + } + fn support_test_catalog_xml(uuid: &str) -> String { format!( r#" diff --git a/crates/unica-coder/src/infrastructure/native_operations/support.rs b/crates/unica-coder/src/infrastructure/native_operations/support.rs index c872bbce..2cfa6388 100644 --- a/crates/unica-coder/src/infrastructure/native_operations/support.rs +++ b/crates/unica-coder/src/infrastructure/native_operations/support.rs @@ -8,8 +8,9 @@ use std::path::{Path, PathBuf}; use super::common::{ absolutize, find_support_config_dir, guard_active_format_dependencies, - guard_exact_preimage_if_unprotected, is_uuid_text, parse_support_header, path_arg, - support_root_uuid_from_bytes, support_uuid_dependency_paths, MutationData, + guard_exact_preimage_if_unprotected, is_uuid_text, parse_quoted_support_strings, + parse_support_header, path_arg, support_root_uuid_from_bytes, support_uuid_dependency_paths, + MutationData, }; use super::compile_transaction::{ CompileTransaction, DirectoryMembershipSelector, DirectoryMembershipSnapshot, @@ -47,6 +48,13 @@ impl SupportCapability { fn enabled(self) -> bool { matches!(self, Self::On) } + + fn argument_value(self) -> &'static str { + match self { + Self::On => "on", + Self::Off => "off", + } + } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -244,6 +252,7 @@ fn edit_support_execution( )); } + let mut capability_vendor_payload_preimages = None; let (mut outcome, updated, data) = match action { SupportEditAction::Capability(capability) => { if (global_flag == 0) == capability.enabled() { @@ -257,6 +266,18 @@ fn edit_support_execution( format!("Возможность изменения конфигурации уже {word} — изменений нет."), )); } + let vendor_payload_preimages = support_vendor_payload_preimages(&config_dir)?; + preflight_required_vendor_payloads( + &config_dir, + &text, + vendor_count, + capability, + vendor_payload_preimages + .1 + .iter() + .map(|(path, _)| path.as_path()), + )?; + capability_vendor_payload_preimages = Some(vendor_payload_preimages); plan_capability(&bin_path, &text, capability, &resolved_path) } SupportEditAction::Set(rule) => { @@ -296,7 +317,11 @@ fn edit_support_execution( } let (vendor_payload_snapshot, vendor_payload_reads) = - support_vendor_payload_preimages(&config_dir)?; + if let Some(preimages) = capability_vendor_payload_preimages { + preimages + } else { + support_vendor_payload_preimages(&config_dir)? + }; let updated_bytes = parent_configurations_bytes(&updated); let mut transaction = CompileTransaction::new(); transaction.replace_bytes(&bin_path, &raw, updated_bytes.clone())?; @@ -370,6 +395,57 @@ pub(crate) fn support_edit_reads_uuid_dependency(args: &Map) -> b matches!(support_edit_action(args), Ok(SupportEditAction::Set(_))) } +/// Read-only capability preflight used by the public application guard. +/// +/// The mutation handler repeats this validation while binding exact preimages. +/// This keeps preview and apply aligned without weakening commit-time race +/// protection. +pub(crate) fn preflight_support_edit_capability( + args: &Map, + context: &WorkspaceContext, +) -> Result<(), String> { + let SupportEditAction::Capability(capability) = support_edit_action(args)? else { + return Ok(()); + }; + let target_path = support_target_path(args, context)?; + if !target_path.exists() { + return Err(format!("Путь не найден: {}", target_path.display())); + } + let resolved_path = target_path + .canonicalize() + .unwrap_or_else(|_| target_path.clone()); + let Some(config_dir) = find_support_config_dir(&resolved_path) else { + return Err(format!( + "Не найден корень конфигурации (Configuration.xml) над путём: {}", + resolved_path.display() + )); + }; + let bin_path = config_dir.join("Ext").join("ParentConfigurations.bin"); + if !bin_path.exists() { + return Ok(()); + } + let raw = fs::read(&bin_path) + .map_err(|error| format!("failed to read {}: {error}", bin_path.display()))?; + if raw.len() <= 32 { + return Ok(()); + } + let text = decode_parent_configurations(&raw)?; + let Some((global_flag, vendor_count)) = parse_support_header(&text) else { + return Err("Неизвестный формат ParentConfigurations.bin".to_string()); + }; + if vendor_count == 0 || (global_flag == 0) == capability.enabled() { + return Ok(()); + } + let (_, vendor_payloads) = support_vendor_payload_paths(&config_dir)?; + preflight_required_vendor_payloads( + &config_dir, + &text, + vendor_count, + capability, + vendor_payloads.iter().map(PathBuf::as_path), + ) +} + fn string_arg(args: &Map, names: &[&str]) -> Option { names .iter() @@ -539,6 +615,25 @@ fn parent_configurations_bytes(text: &str) -> Vec { fn support_vendor_payload_preimages( config_dir: &Path, ) -> Result { + let (snapshot, paths) = support_vendor_payload_paths(config_dir)?; + let reads = paths + .into_iter() + .map(|path| { + let preimage = fs::read(&path).map_err(|error| { + format!( + "failed to read support vendor payload {}: {error}", + path.display() + ) + })?; + Ok::<_, String>((path, preimage)) + }) + .collect::, _>>()?; + Ok((snapshot, reads)) +} + +fn support_vendor_payload_paths( + config_dir: &Path, +) -> Result<(DirectoryMembershipSnapshot, Vec), String> { let directory = config_dir.join("Ext").join("ParentConfigurations"); let metadata = match fs::symlink_metadata(&directory) { Ok(metadata) => metadata, @@ -601,37 +696,115 @@ fn support_vendor_payload_preimages( }) .collect(), ); - let reads = paths - .into_iter() - .map(|path| { - let metadata = fs::symlink_metadata(&path).map_err(|error| { - format!( - "failed to inspect support vendor payload {}: {error}", - path.display() - ) - })?; - if metadata_is_link_or_reparse_point(&metadata) { - return Err(format!( - "support vendor payload must not be a symbolic link or reparse point: {}", - path.display() - )); - } - if !metadata.is_file() { - return Err(format!( - "support vendor payload is not a regular file: {}", - path.display() - )); - } - let preimage = fs::read(&path).map_err(|error| { - format!( - "failed to read support vendor payload {}: {error}", - path.display() - ) - })?; - Ok((path, preimage)) + for path in &paths { + let metadata = fs::symlink_metadata(path).map_err(|error| { + format!( + "failed to inspect support vendor payload {}: {error}", + path.display() + ) + })?; + if metadata_is_link_or_reparse_point(&metadata) { + return Err(format!( + "support vendor payload must not be a symbolic link or reparse point: {}", + path.display() + )); + } + if !metadata.is_file() { + return Err(format!( + "support vendor payload is not a regular file: {}", + path.display() + )); + } + fs::File::open(path).map_err(|error| { + format!( + "failed to open support vendor payload {} for reading: {error}", + path.display() + ) + })?; + } + Ok((snapshot, paths)) +} + +fn preflight_required_vendor_payloads<'a>( + config_dir: &Path, + parent_configurations: &str, + vendor_count: usize, + capability: SupportCapability, + vendor_payloads: impl IntoIterator, +) -> Result<(), String> { + let quoted = parse_quoted_support_strings(parent_configurations); + let expected_fields = vendor_count.checked_mul(3).ok_or_else(|| { + "Слишком много записей поставщиков в ParentConfigurations.bin".to_string() + })?; + if quoted.len() != expected_fields { + return Err(format!( + "Не удалось сопоставить записи поставщиков в ParentConfigurations.bin: заголовок объявляет {vendor_count}, строковых полей {}", + quoted.len() + )); + } + support_vendor_rule_flags(parent_configurations, vendor_count)?; + let vendor_payloads = vendor_payloads.into_iter().collect::>(); + let missing = quoted + .chunks_exact(3) + .map(|fields| fields[2].as_str()) + .filter(|vendor_name| { + !vendor_payloads.iter().any(|path| { + path.file_stem() + .and_then(|stem| stem.to_str()) + .is_some_and(|stem| stem == *vendor_name) + && path + .extension() + .and_then(|extension| extension.to_str()) + .is_some_and(|extension| extension.eq_ignore_ascii_case("cf")) + }) }) - .collect::, _>>()?; - Ok((snapshot, reads)) + .map(|vendor_name| { + config_dir + .join("Ext") + .join("ParentConfigurations") + .join(format!("{vendor_name}.cf")) + }) + .collect::>(); + if missing.is_empty() { + return Ok(()); + } + + Err(format!( + "Capability={} cannot be applied: missing required vendor payload(s): {}. ParentConfigurations.bin was not changed; export the matching vendor configuration payload with the 1C platform first.", + capability.argument_value(), + missing + .iter() + .map(|path| path.display().to_string()) + .collect::>() + .join(", ") + )) +} + +fn support_vendor_rule_flags(text: &str, expected_count: usize) -> Result, String> { + let mut flags = Vec::with_capacity(expected_count); + let mut cursor = 0usize; + while cursor < text.len() { + if let Some((flag_start, flag_end)) = vendor_flag_span(text, cursor) { + let flag = text[flag_start..flag_end].parse::().map_err(|_| { + "Неизвестный флаг поставщика в ParentConfigurations.bin".to_string() + })?; + flags.push(flag); + cursor = flag_end; + continue; + } + let ch = text[cursor..] + .chars() + .next() + .expect("cursor remains on a valid character boundary"); + cursor += ch.len_utf8(); + } + if flags.len() != expected_count { + return Err(format!( + "Не удалось сопоставить флаги поставщиков в ParentConfigurations.bin: заголовок объявляет {expected_count}, разобрано {}", + flags.len() + )); + } + Ok(flags) } fn replace_global_flag(text: &str, target: u8) -> Result { @@ -775,6 +948,9 @@ mod tests { fs::write(&config_path, configuration_xml(version)).unwrap(); let bin_path = source.join("Ext/ParentConfigurations.bin"); fs::write(&bin_path, parent_configurations()).unwrap(); + let vendor_dir = source.join("Ext/ParentConfigurations"); + fs::create_dir(&vendor_dir).unwrap(); + fs::write(vendor_dir.join("VendorConf.cf"), b"platform vendor payload").unwrap(); let context = WorkspaceContext { cwd: root.clone(), workspace_root: root.clone(), @@ -873,6 +1049,34 @@ mod tests { .to_string() } + fn locked_parent_configurations_without_payload() -> String { + parent_configurations() + .replacen("{6,0,1,", "{6,1,1,", 1) + .replacen( + "dddddddd-dddd-dddd-dddd-dddddddddddd,0,eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee", + "dddddddd-dddd-dddd-dddd-dddddddddddd,1,eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee", + 1, + ) + } + + fn editable_parent_configurations_with_unlocked_vendor_without_payload() -> String { + parent_configurations().replacen( + "dddddddd-dddd-dddd-dddd-dddddddddddd,0,eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee", + "dddddddd-dddd-dddd-dddd-dddddddddddd,1,eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee", + 1, + ) + } + + fn editable_parent_configurations_with_two_vendors() -> String { + parent_configurations() + .replacen("{6,0,1,", "{6,0,2,", 1) + .replacen( + ",3,1,0,aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + ",ffffffff-ffff-ffff-ffff-ffffffffffff,0,cccccccc-cccc-cccc-cccc-cccccccccccc,\"2.0\",\"Second Vendor\",\"SecondVendor\",3,1,0,aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + 1, + ) + } + #[test] fn support_edit_replaces_parent_configurations_transactionally() { let fixture = SupportFixture::new("transaction", "2.20"); @@ -890,7 +1094,7 @@ mod tests { fn support_capability_off_only_changes_global_flag_and_locks_object_rules() { let fixture = SupportFixture::new("capability-off-semantics", "2.20"); let vendor_dir = fixture.root.join("src/Ext/ParentConfigurations"); - fs::create_dir(&vendor_dir).unwrap(); + fs::create_dir_all(&vendor_dir).unwrap(); let vendor_payload = vendor_dir.join("VendorConf.cf"); let vendor_bytes = b"platform vendor payload".to_vec(); fs::write(&vendor_payload, &vendor_bytes).unwrap(); @@ -917,6 +1121,131 @@ mod tests { assert_eq!(fs::read(vendor_payload).unwrap(), vendor_bytes); } + #[test] + fn support_capability_on_rejects_missing_required_vendor_payload_without_writing_bin() { + let fixture = SupportFixture::new("capability-on-missing-vendor-payload", "2.20"); + fs::remove_dir_all(fixture.root.join("src/Ext/ParentConfigurations")).unwrap(); + fs::write( + &fixture.bin_path, + locked_parent_configurations_without_payload(), + ) + .unwrap(); + let bin_before = fs::read(&fixture.bin_path).unwrap(); + let args = json!({ + "Path": "src", + "Capability": "on" + }) + .as_object() + .unwrap() + .clone(); + + let error = edit_support_result(&args, &fixture.context).unwrap_err(); + + assert!(error.contains("Capability=on"), "{error}"); + assert!(error.contains("required vendor payload"), "{error}"); + assert!(error.contains("VendorConf.cf"), "{error}"); + assert_eq!(fs::read(&fixture.bin_path).unwrap(), bin_before); + } + + #[test] + fn support_capability_off_rejects_missing_required_vendor_payload_without_writing_bin() { + let fixture = SupportFixture::new("capability-off-missing-vendor-payload", "2.20"); + fs::remove_dir_all(fixture.root.join("src/Ext/ParentConfigurations")).unwrap(); + fs::write( + &fixture.bin_path, + editable_parent_configurations_with_unlocked_vendor_without_payload(), + ) + .unwrap(); + let bin_before = fs::read(&fixture.bin_path).unwrap(); + let args = json!({ + "Path": "src", + "Capability": "off" + }) + .as_object() + .unwrap() + .clone(); + + let error = edit_support_result(&args, &fixture.context).unwrap_err(); + + assert!(error.contains("Capability=off"), "{error}"); + assert!(error.contains("required vendor payload"), "{error}"); + assert!(error.contains("VendorConf.cf"), "{error}"); + assert_eq!(fs::read(&fixture.bin_path).unwrap(), bin_before); + } + + #[test] + fn support_capability_off_rejects_missing_payload_for_already_locked_vendor() { + let fixture = SupportFixture::new("capability-off-locked-vendor-missing-payload", "2.20"); + fs::remove_dir_all(fixture.root.join("src/Ext/ParentConfigurations")).unwrap(); + let bin_before = fs::read(&fixture.bin_path).unwrap(); + + let error = + edit_support_result(&fixture.capability_off_args(), &fixture.context).unwrap_err(); + + assert!(error.contains("Capability=off"), "{error}"); + assert!(error.contains("VendorConf.cf"), "{error}"); + assert_eq!(fs::read(&fixture.bin_path).unwrap(), bin_before); + } + + #[test] + fn support_capability_off_validates_every_vendor_in_the_post_image() { + let fixture = SupportFixture::new("capability-off-multi-vendor-missing-payload", "2.20"); + fs::write( + &fixture.bin_path, + editable_parent_configurations_with_two_vendors(), + ) + .unwrap(); + let vendor_dir = fixture.root.join("src/Ext/ParentConfigurations"); + fs::remove_dir_all(&vendor_dir).unwrap(); + fs::create_dir(&vendor_dir).unwrap(); + fs::write(vendor_dir.join("VendorConf.cf"), b"first vendor payload").unwrap(); + let bin_before = fs::read(&fixture.bin_path).unwrap(); + + let error = + edit_support_result(&fixture.capability_off_args(), &fixture.context).unwrap_err(); + + assert!(error.contains("Capability=off"), "{error}"); + assert!(error.contains("SecondVendor.cf"), "{error}"); + assert!(!error.contains("VendorConf.cf,"), "{error}"); + assert_eq!(fs::read(&fixture.bin_path).unwrap(), bin_before); + } + + #[test] + fn support_capability_on_accepts_and_preserves_matching_vendor_payload() { + let fixture = SupportFixture::new("capability-on-with-vendor-payload", "2.20"); + fs::write( + &fixture.bin_path, + locked_parent_configurations_without_payload(), + ) + .unwrap(); + let vendor_dir = fixture.root.join("src/Ext/ParentConfigurations"); + fs::remove_dir_all(&vendor_dir).unwrap(); + fs::create_dir(&vendor_dir).unwrap(); + let vendor_payload = vendor_dir.join("VendorConf.CF"); + let vendor_bytes = b"platform vendor payload".to_vec(); + fs::write(&vendor_payload, &vendor_bytes).unwrap(); + let args = json!({ + "Path": "src", + "Capability": "on" + }) + .as_object() + .unwrap() + .clone(); + + let outcome = edit_support_result(&args, &fixture.context).unwrap(); + + assert!(outcome.ok, "{outcome:?}"); + let updated = fs::read_to_string(&fixture.bin_path).unwrap(); + assert!(updated.contains("{6,0,1,"), "{updated}"); + assert!( + updated.contains( + "dddddddd-dddd-dddd-dddd-dddddddddddd,0,eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee" + ), + "{updated}" + ); + assert_eq!(fs::read(vendor_payload).unwrap(), vendor_bytes); + } + #[test] fn support_edit_rejects_a_concurrent_parent_configurations_change() { let fixture = SupportFixture::new("bin-race", "2.20"); @@ -942,7 +1271,7 @@ mod tests { fn support_edit_rejects_a_concurrent_vendor_payload_change() { let fixture = SupportFixture::new("vendor-payload-race", "2.20"); let vendor_dir = fixture.root.join("src/Ext/ParentConfigurations"); - fs::create_dir(&vendor_dir).unwrap(); + fs::create_dir_all(&vendor_dir).unwrap(); let vendor_payload = vendor_dir.join("VendorConf.cf"); let original_vendor = b"original vendor payload".to_vec(); let concurrent_vendor = b"concurrent vendor payload".to_vec(); @@ -966,7 +1295,7 @@ mod tests { fn support_edit_rejects_a_concurrent_vendor_payload_create_case_insensitively() { let fixture = SupportFixture::new("vendor-payload-create-race", "2.20"); let vendor_dir = fixture.root.join("src/Ext/ParentConfigurations"); - fs::create_dir(&vendor_dir).unwrap(); + fs::create_dir_all(&vendor_dir).unwrap(); let existing_payload = vendor_dir.join("VendorConf.cf"); fs::write(&existing_payload, b"original vendor payload").unwrap(); let concurrent_payload = vendor_dir.join("ConcurrentVendor.CF"); diff --git a/crates/unica-coder/src/infrastructure/support_guard.rs b/crates/unica-coder/src/infrastructure/support_guard.rs index 52fdbe15..00b31e74 100644 --- a/crates/unica-coder/src/infrastructure/support_guard.rs +++ b/crates/unica-coder/src/infrastructure/support_guard.rs @@ -10,6 +10,7 @@ use crate::infrastructure::native_operations::common::{ }; use crate::infrastructure::native_operations::compile_transaction::CompileTransaction; use crate::infrastructure::native_operations::role::resolve_role_edit_guard_path; +use crate::infrastructure::native_operations::support; use crate::infrastructure::native_operations::template; use crate::infrastructure::native_operations::xdto::resolve_xdto_guard_path; use crate::infrastructure::source_roots::normalize_path_identity; @@ -34,6 +35,27 @@ pub(crate) fn evaluate_support_guard( args: &Map, context: &WorkspaceContext, ) -> Result { + if matches!( + spec.handler, + ToolHandler::NativeOperation { + operation: "support-edit", + .. + } + ) { + if let Err(error) = support::preflight_support_edit_capability(args, context) { + return Ok(SupportGuardCheck::Block(AdapterOutcome { + ok: false, + summary: "support-edit failed".to_string(), + changes: Vec::new(), + warnings: Vec::new(), + errors: vec![error], + artifacts: Vec::new(), + stdout: None, + stderr: None, + command: None, + })); + } + } let Some((target_path, requirement)) = support_guard_target(spec, args, context) else { return Ok(SupportGuardCheck::Allow); }; diff --git a/scripts/dev/verify-issue-76-roundtrip.py b/scripts/dev/verify-issue-76-roundtrip.py new file mode 100644 index 00000000..ea583ea9 --- /dev/null +++ b/scripts/dev/verify-issue-76-roundtrip.py @@ -0,0 +1,2580 @@ +#!/usr/bin/env python3 +"""Run the opt-in live round-trip reproducer for Unica issue #76. + +The verifier never points Unica or 1C at the supplied inputs. It first copies +the file infobase and the Designer source tree into a fresh private workspace, +then performs every mutation and runtime operation through the public Unica MCP +surface. This is developer evidence, not a product or release entry point. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import os +import queue +import re +import secrets +import shlex +import signal +import stat +import subprocess +import sys +import tempfile +import threading +import time +import xml.etree.ElementTree as ElementTree +from collections import deque +from pathlib import Path + + +SCENARIO = "issue-76-live-roundtrip" +SOURCE_SET = "main" +CATALOG_METADATA_PATH = "Catalog.ЗависимостиСчетов" +MODULE_METADATA_PATH = ( + "CommonModule.СообщенияВСлужбуТехническойПоддержкиБПКлиентСервер.Module" +) +CATALOG_RELATIVE_PATH = Path("src/Catalogs/ЗависимостиСчетов.xml") +COMMON_MODULE_DESCRIPTOR_RELATIVE_PATH = Path( + "src/CommonModules/" + "СообщенияВСлужбуТехническойПоддержкиБПКлиентСервер.xml" +) +MODULE_RELATIVE_PATH = Path( + "src/CommonModules/" + "СообщенияВСлужбуТехническойПоддержкиБПКлиентСервер/Ext/Module.bsl" +) +PARENT_CONFIGURATION_RELATIVE_PATH = Path( + "src/Ext/ParentConfigurations/УправлениеХолдингом.cf" +) +CONFIG_DUMP_INFO_RELATIVE_PATH = Path("src/ConfigDumpInfo.xml") +MARKER_PREFIX = "UNICA_ISSUE_76_ROUND_TRIP" +MARKER_RE = re.compile(rf"{MARKER_PREFIX}(?:_[0-9A-F]{{32}})?\Z") +EDITABLE_SUPPORT_RULE_RECEIPTS = frozenset( + { + "editable", + "редактируется с сохранением поддержки " + "(объект продолжит получать обновления вендора — возможны конфликты при обновлении)", + } +) +DEFAULT_TIMEOUT_SECONDS = 7200.0 +MAX_TIMEOUT_SECONDS = 86400.0 +DIAGNOSTIC_LIMIT = 4096 +REQUIRED_TOOLS = frozenset( + { + "unica.cf.info", + "unica.support.edit", + "unica.meta.edit", + "unica.code.patch", + "unica.runtime.execute", + } +) +PLATFORM_VERSION_RE = re.compile(r"8\.3\.27\.\d+\Z") +_CONNECTION_CREDENTIAL_RE = re.compile( + r"(?i)(? float: + try: + parsed = float(value) + except ValueError as error: + raise argparse.ArgumentTypeError("must be a number") from error + if ( + not math.isfinite(parsed) + or parsed <= 0 + or parsed > MAX_TIMEOUT_SECONDS + ): + raise argparse.ArgumentTypeError( + f"must be finite and in the range 0 < seconds <= {MAX_TIMEOUT_SECONDS:g}" + ) + return parsed + + +def _platform_version(value: str) -> str: + if PLATFORM_VERSION_RE.fullmatch(value) is None: + raise argparse.ArgumentTypeError("must be an exact 8.3.27.x platform version") + return value + + +def _argument_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=( + "Copy a file infobase and Designer source tree, mutate the copies " + "through packaged Unica, build them, and verify a safe full dump." + ) + ) + parser.add_argument("--binary", required=True, type=Path) + parser.add_argument("--binary-arg", action="append", default=[]) + parser.add_argument( + "--plugin-root", + required=True, + type=Path, + help="single-target packaged Unica runtime root", + ) + parser.add_argument("--database", required=True, type=Path) + parser.add_argument("--sources", required=True, type=Path) + parser.add_argument( + "--parent-configuration", + required=True, + type=Path, + help=( + "exact УправлениеХолдингом.cf vendor payload copied only into the " + "private source tree" + ), + ) + parser.add_argument("--platform-path", required=True, type=Path) + parser.add_argument("--platform-version", required=True, type=_platform_version) + parser.add_argument("--report", required=True, type=Path) + parser.add_argument("--evidence-dir", type=Path) + parser.add_argument( + "--builder", + choices=("DESIGNER", "IBCMD"), + default="DESIGNER", + ) + parser.add_argument("--db-user", default="Администратор") + parser.add_argument( + "--timeout-seconds", + type=_positive_timeout, + default=DEFAULT_TIMEOUT_SECONDS, + ) + parser.add_argument( + "--execute", + action="store_true", + required=True, + help="explicitly opt in to running the mutating live scenario on copies", + ) + parser.add_argument( + "--allow-empty-password", + action="store_true", + required=True, + help="explicitly confirm the selected copied infobase uses an empty password", + ) + return parser + + +def _is_relative_to(path: Path, parent: Path) -> bool: + try: + path.relative_to(parent) + except ValueError: + return False + return True + + +def _paths_overlap(left: Path, right: Path) -> bool: + return _is_relative_to(left, right) or _is_relative_to(right, left) + + +def _resolved_absolute(path: Path, label: str, *, directory: bool) -> Path: + path = Path(path) + if not path.is_absolute(): + raise SourceError(f"{label} must be an absolute path") + if path.is_symlink(): + raise SourceError(f"{label} must not be a symlink") + try: + resolved = path.resolve(strict=True) + except OSError as error: + raise SourceError(f"{label} does not exist: {error}") from error + if directory and not resolved.is_dir(): + raise SourceError(f"{label} must be a directory") + if not directory and not resolved.is_file(): + raise SourceError(f"{label} must be a regular file") + return resolved + + +def _validate_report_path( + report_path: Path, + *, + protected_paths: tuple[tuple[Path, str], ...], +) -> Path: + raw = Path(report_path) + if not raw.is_absolute(): + raise SourceError("report path must be absolute") + if raw.is_symlink(): + raise SourceError("report path must not be a symlink") + try: + parent = raw.parent.resolve(strict=True) + except OSError as error: + raise SourceError(f"report parent directory does not exist: {error}") from error + if not parent.is_dir(): + raise SourceError("report parent must be a directory") + target = parent / raw.name + if target.exists() and (target.is_symlink() or not target.is_file()): + raise SourceError("report target exists and is not a safe regular file") + for protected, label in protected_paths: + if _paths_overlap(target, protected): + raise SourceError(f"report path must be outside {label}") + return target + + +def _validate_evidence_directory( + evidence_dir: Path, + *, + database: Path, + sources: Path, + report_path: Path, + protected_paths: tuple[tuple[Path, str], ...] = (), +) -> Path: + evidence = _resolved_absolute(evidence_dir, "evidence directory", directory=True) + repo = Path(__file__).resolve().parents[2] + home = Path.home().resolve() + filesystem_root = Path(evidence.anchor).resolve() + if ( + evidence == filesystem_root + or evidence == home + or _paths_overlap(evidence, repo) + or _paths_overlap(evidence, database) + or _paths_overlap(evidence, sources) + or any(_paths_overlap(evidence, path) for path, _label in protected_paths) + or _is_relative_to(home, evidence) + or _paths_overlap(evidence, report_path) + ): + raise SourceError( + "evidence directory must be outside broad, home, repository, input, " + "runtime, and report paths" + ) + try: + with os.scandir(evidence) as entries: + if next(entries, None) is not None: + raise SourceError("evidence directory must be empty") + except OSError as error: + raise SourceError(f"cannot inspect evidence directory: {error}") from error + return evidence + + +def _safe_automatic_evidence_parent( + *, + protected_paths: tuple[tuple[Path, str], ...], +) -> Path: + raw_candidates = [ + tempfile.tempdir, + os.environ.get("TMPDIR"), + os.environ.get("TEMP"), + os.environ.get("TMP"), + "/private/tmp", + "/tmp", + "/var/tmp", + "/usr/tmp", + ] + seen = set() + home = Path.home().resolve() + repo = Path(__file__).resolve().parents[2] + for raw in raw_candidates: + if raw in (None, "", b""): + continue + try: + candidate = Path(os.fsdecode(raw)) + except (TypeError, ValueError): + continue + if not candidate.is_absolute(): + continue + try: + candidate = candidate.resolve(strict=True) + except OSError: + continue + if not candidate.is_dir() or candidate == Path(candidate.anchor): + continue + if _is_relative_to(candidate, home) or _is_relative_to(candidate, repo): + continue + if any( + _is_relative_to(candidate, protected) + for protected, _label in protected_paths + ): + continue + if not os.access(candidate, os.W_OK | os.X_OK): + continue + seen_key = os.path.normcase(str(candidate)) + if seen_key in seen: + continue + seen.add(seen_key) + return candidate + raise SourceError( + "no safe automatic evidence parent is available outside protected paths" + ) + + +def _require_regular_file(path: Path, label: str) -> Path: + try: + metadata = path.lstat() + except OSError as error: + raise SourceError(f"cannot inspect {label}: {error}") from error + reparse_flag = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0) + file_attributes = getattr(metadata, "st_file_attributes", 0) + if ( + not stat.S_ISREG(metadata.st_mode) + or stat.S_ISLNK(metadata.st_mode) + or bool(reparse_flag and file_attributes & reparse_flag) + ): + raise SourceError(f"{label} must be a regular non-symlink file: {path}") + return path + + +def _require_unlinked_directory_ancestors(path: Path, label: str) -> None: + reparse_flag = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0) + for candidate in (path.parent, *path.parent.parents): + try: + metadata = candidate.lstat() + except OSError as error: + raise SourceError(f"cannot inspect {label} ancestor {candidate}: {error}") from error + file_attributes = getattr(metadata, "st_file_attributes", 0) + if ( + not stat.S_ISDIR(metadata.st_mode) + or stat.S_ISLNK(metadata.st_mode) + or bool(reparse_flag and file_attributes & reparse_flag) + ): + raise SourceError( + f"{label} ancestor must be a real directory, not a link or reparse point: " + f"{candidate}" + ) + + +def _hash_file(path: Path) -> str: + digest = hashlib.sha256() + try: + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + except OSError as error: + raise SourceError(f"cannot hash regular file {path}: {error}") from error + return digest.hexdigest() + + +def _copy_regular_file(source: Path, destination: Path) -> tuple[str, int]: + try: + before = source.lstat() + except OSError as error: + raise SourceError(f"cannot inspect input file {source}: {error}") from error + if not stat.S_ISREG(before.st_mode): + raise SourceError(f"input entry is not a regular file: {source}") + if before.st_nlink != 1: + raise SourceError( + f"input file has a hardlink alias and cannot be copied safely: {source}" + ) + source_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + destination_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + mode = 0o700 if before.st_mode & stat.S_IXUSR else 0o600 + digest = hashlib.sha256() + total = 0 + try: + source_fd = os.open(source, source_flags) + try: + opened = os.fstat(source_fd) + if ( + not stat.S_ISREG(opened.st_mode) + or opened.st_dev != before.st_dev + or opened.st_ino != before.st_ino + ): + raise SourceError(f"input file changed before it could be copied: {source}") + destination_fd = os.open(destination, destination_flags, mode) + try: + while True: + block = os.read(source_fd, 1024 * 1024) + if not block: + break + digest.update(block) + total += len(block) + view = memoryview(block) + while view: + written = os.write(destination_fd, view) + view = view[written:] + os.fchmod(destination_fd, mode) + finally: + os.close(destination_fd) + after = os.fstat(source_fd) + if ( + after.st_dev != before.st_dev + or after.st_ino != before.st_ino + or after.st_size != before.st_size + or after.st_mtime_ns != before.st_mtime_ns + ): + raise SourceError(f"input file changed while it was copied: {source}") + finally: + os.close(source_fd) + except SourceError: + raise + except OSError as error: + raise SourceError(f"cannot make private copy of {source}: {error}") from error + return digest.hexdigest(), total + + +def _copy_regular_tree(source: Path, destination: Path) -> dict: + """Copy a tree without links/special entries and return a compact receipt.""" + + if destination.exists() or destination.is_symlink(): + raise SourceError(f"private copy target already exists: {destination}") + try: + destination.mkdir(mode=0o700) + except OSError as error: + raise SourceError(f"cannot create private copy target: {error}") from error + + receipt_digest = hashlib.sha256() + file_count = 0 + directory_count = 1 + total_bytes = 0 + identities: set[tuple[int, int]] = set() + + def visit(source_dir: Path, destination_dir: Path, relative: Path) -> None: + nonlocal file_count, directory_count, total_bytes + try: + with os.scandir(source_dir) as iterator: + entries = sorted(iterator, key=lambda entry: entry.name) + except OSError as error: + raise SourceError(f"cannot enumerate input tree {source_dir}: {error}") from error + if not entries: + receipt_digest.update(b"D\0" + relative.as_posix().encode("utf-8") + b"\0") + for entry in entries: + source_path = Path(entry.path) + destination_path = destination_dir / entry.name + child_relative = relative / entry.name + try: + if entry.is_symlink(): + raise SourceError(f"input symlink is forbidden: {source_path}") + metadata = entry.stat(follow_symlinks=False) + identity = (metadata.st_dev, metadata.st_ino) + if identity in identities: + raise SourceError( + f"input filesystem identity is exposed more than once: {source_path}" + ) + identities.add(identity) + if entry.is_dir(follow_symlinks=False): + destination_path.mkdir(mode=0o700) + directory_count += 1 + receipt_digest.update( + b"D\0" + child_relative.as_posix().encode("utf-8") + b"\0" + ) + visit(source_path, destination_path, child_relative) + elif entry.is_file(follow_symlinks=False): + file_sha256, size = _copy_regular_file(source_path, destination_path) + file_count += 1 + total_bytes += size + receipt_digest.update( + b"F\0" + + child_relative.as_posix().encode("utf-8") + + b"\0" + + file_sha256.encode("ascii") + + b"\0" + ) + else: + raise SourceError(f"special input entry is forbidden: {source_path}") + except SourceError: + raise + except OSError as error: + raise SourceError(f"cannot copy input entry {source_path}: {error}") from error + + visit(source, destination, Path()) + return { + "sha256": receipt_digest.hexdigest(), + "fileCount": file_count, + "directoryCount": directory_count, + "bytes": total_bytes, + } + + +def _install_parent_configuration(parent_configuration: Path, source_copy: Path) -> dict: + """Copy the explicit vendor payload into only one already-private source tree.""" + + parent_configuration = _require_regular_file( + Path(parent_configuration), + "parent configuration payload", + ) + source_copy = Path(source_copy) + if source_copy.is_symlink() or not source_copy.is_dir(): + raise SourceError("private source copy must be a regular directory") + destination = ( + source_copy / PARENT_CONFIGURATION_RELATIVE_PATH.relative_to("src") + ) + if destination.exists() or destination.is_symlink(): + raise SourceError( + "private source copy already contains the parent configuration payload; " + "refusing to overwrite or merge it" + ) + directory = destination.parent + if directory.exists(): + if directory.is_symlink() or not directory.is_dir(): + raise SourceError( + "private parent configuration destination must be a regular directory" + ) + else: + try: + directory.mkdir(mode=0o700) + except OSError as error: + raise SourceError( + f"cannot create private parent configuration directory: {error}" + ) from error + sha256, size = _copy_regular_file(parent_configuration, destination) + return { + "sha256": sha256, + "bytes": size, + "destination": "$EVIDENCE/workspace/" + + PARENT_CONFIGURATION_RELATIVE_PATH.as_posix(), + } + + +def _regular_file_stat_signature(path: Path) -> tuple[int, int, int, int, int, int]: + path = _require_regular_file(path, "input file") + try: + metadata = path.stat() + except OSError as error: + raise SourceError(f"cannot inspect input file {path}: {error}") from error + return ( + metadata.st_dev, + metadata.st_ino, + metadata.st_mode, + metadata.st_size, + metadata.st_mtime_ns, + metadata.st_ctime_ns, + ) + + +def _stat_tree_digest(root: Path) -> str: + """Bind input identities and timestamps without reading all bytes again.""" + + digest = hashlib.sha256() + identities: set[tuple[int, int]] = set() + try: + root_metadata = root.lstat() + except OSError as error: + raise SourceError(f"cannot inspect input state {root}: {error}") from error + if not stat.S_ISDIR(root_metadata.st_mode): + raise SourceError(f"input state root is not a directory: {root}") + identities.add((root_metadata.st_dev, root_metadata.st_ino)) + digest.update( + b"R\0.\0" + + str(root_metadata.st_mode).encode("ascii") + + b"\0" + + str(root_metadata.st_size).encode("ascii") + + b"\0" + + str(root_metadata.st_mtime_ns).encode("ascii") + + b"\0" + + str(root_metadata.st_ctime_ns).encode("ascii") + + b"\0" + + str(root_metadata.st_dev).encode("ascii") + + b"\0" + + str(root_metadata.st_ino).encode("ascii") + + b"\0" + ) + + def visit(directory: Path, relative: Path) -> None: + try: + with os.scandir(directory) as iterator: + entries = sorted(iterator, key=lambda entry: entry.name) + except OSError as error: + raise SourceError(f"cannot enumerate input state {directory}: {error}") from error + for entry in entries: + path = Path(entry.path) + if entry.is_symlink(): + raise SourceError(f"input symlink is forbidden: {path}") + try: + metadata = entry.stat(follow_symlinks=False) + except OSError as error: + raise SourceError(f"cannot inspect input state {path}: {error}") from error + identity = (metadata.st_dev, metadata.st_ino) + if identity in identities: + raise SourceError(f"input hardlink alias is forbidden: {path}") + identities.add(identity) + child = relative / entry.name + kind = b"D" if entry.is_dir(follow_symlinks=False) else b"F" + if kind == b"F" and not entry.is_file(follow_symlinks=False): + raise SourceError(f"special input entry is forbidden: {path}") + record = ( + kind + + b"\0" + + child.as_posix().encode("utf-8") + + b"\0" + + str(metadata.st_mode).encode("ascii") + + b"\0" + + str(metadata.st_size).encode("ascii") + + b"\0" + + str(metadata.st_mtime_ns).encode("ascii") + + b"\0" + + str(metadata.st_ctime_ns).encode("ascii") + + b"\0" + + str(metadata.st_dev).encode("ascii") + + b"\0" + + str(metadata.st_ino).encode("ascii") + + b"\0" + ) + digest.update(record) + if kind == b"D": + visit(path, child) + + visit(root, Path()) + return digest.hexdigest() + + +def _write_new_file(path: Path, payload: bytes, *, mode: int) -> None: + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + try: + descriptor = os.open(path, flags, mode) + try: + view = memoryview(payload) + while view: + written = os.write(descriptor, view) + view = view[written:] + os.fchmod(descriptor, mode) + os.fsync(descriptor) + finally: + os.close(descriptor) + except OSError as error: + raise SourceError(f"cannot create private project file {path}: {error}") from error + + +def _yaml_string(value: str) -> str: + if not isinstance(value, str) or not value or "\0" in value: + raise SourceError("project configuration strings must be non-empty and NUL-free") + return json.dumps(value, ensure_ascii=False) + + +def _connection_component(value: str, label: str) -> str: + if not isinstance(value, str) or not value or any(ord(char) < 32 for char in value): + raise SourceError(f"{label} must be a non-empty string without control characters") + return value.replace('"', '""') + + +def _write_project_configuration( + workspace: Path, + *, + database_copy: Path, + platform_path: Path, + platform_version: str, + db_user: str, + builder: str, + timeout_seconds: float, +) -> None: + timeout_ms = max(1, int(round(timeout_seconds * 1000))) + if timeout_ms > 86400000: + raise SourceError("execution timeout exceeds the v8-runner 24-hour limit") + primary = ( + f"execution_timeout: {timeout_ms}\n" + "format: DESIGNER\n" + f"builder: {builder}\n" + "source-set:\n" + f" - name: {SOURCE_SET}\n" + " type: CONFIGURATION\n" + " path: src\n" + ).encode("utf-8") + database_value = _connection_component(str(database_copy), "database copy path") + connection = f'File="{database_value}";' + local = ( + f"workPath: {_yaml_string('work')}\n" + "infobase:\n" + f" connection: {_yaml_string(connection)}\n" + f" user: {_yaml_string(db_user)}\n" + ' password: ""\n' + "tools:\n" + " platform:\n" + f" version: {_yaml_string(platform_version)}\n" + f" path: {_yaml_string(str(platform_path))}\n" + " strict: true\n" + ).encode("utf-8") + _write_new_file(workspace / "v8project.yaml", primary, mode=0o600) + _write_new_file(workspace / "v8project.local.yaml", local, mode=0o600) + + +def _replace_project_platform_path( + workspace: Path, + *, + previous_path: Path, + next_path: Path, +) -> None: + local_path = workspace / "v8project.local.yaml" + _require_regular_file(local_path, "private project local configuration") + try: + preimage = local_path.read_bytes() + except OSError as error: + raise SourceError(f"cannot read private project local configuration: {error}") from error + previous = f" path: {_yaml_string(str(previous_path))}\n".encode("utf-8") + replacement = f" path: {_yaml_string(str(next_path))}\n".encode("utf-8") + if preimage.count(previous) != 1: + raise SourceError( + "private project platform path changed before the full-dump phase" + ) + payload = preimage.replace(previous, replacement, 1) + temporary_path = None + try: + with tempfile.NamedTemporaryFile( + mode="wb", + dir=local_path.parent, + prefix=f".{local_path.name}.issue-76-platform.", + delete=False, + ) as stream: + temporary_path = Path(stream.name) + os.chmod(temporary_path, 0o600) + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + if local_path.read_bytes() != preimage: + raise SourceError( + "private project local configuration changed during platform switch" + ) + os.replace(temporary_path, local_path) + temporary_path = None + except SourceError: + raise + except OSError as error: + raise SourceError( + f"cannot switch private project to the trusted full-dump platform: {error}" + ) from error + finally: + if temporary_path is not None: + try: + temporary_path.unlink(missing_ok=True) + except OSError: + pass + + +def _create_private_ibcmd_platform( + evidence: Path, + *, + trusted_platform: Path, + platform_version: str, +) -> tuple[Path, dict]: + if os.name != "posix": + raise SourceError("private IBCMD data isolation is supported only on POSIX") + trusted_ibcmd = _resolved_absolute( + trusted_platform / "ibcmd", + "trusted platform ibcmd", + directory=False, + ) + if not os.access(trusted_ibcmd, os.X_OK): + raise SourceError("trusted platform ibcmd is not executable") + if PLATFORM_VERSION_RE.fullmatch(platform_version) is None: + raise SourceError("private IBCMD platform requires an exact 8.3.27.x version") + wrapper_root = evidence / "ibcmd-platform" + wrapper_platform = wrapper_root / platform_version + private_data = evidence / "ibcmd-data" + try: + wrapper_root.mkdir(mode=0o700) + wrapper_platform.mkdir(mode=0o700) + private_data.mkdir(mode=0o700) + except OSError as error: + raise SourceError(f"cannot create private IBCMD isolation paths: {error}") from error + wrapper = wrapper_platform / "ibcmd" + payload = ( + "#!/bin/sh\n" + "set -eu\n\n" + 'if [ "${1-}" = "infobase" ]; then\n' + " shift\n" + ' for argument in "$@"; do\n' + ' case "$argument" in\n' + ' --data|--data=*) echo "refusing caller-supplied --data" >&2; exit 64 ;;\n' + " esac\n" + " done\n" + f" exec {shlex.quote(str(trusted_ibcmd))} infobase " + f"--data {shlex.quote(str(private_data))} \"$@\"\n" + "fi\n\n" + f"exec {shlex.quote(str(trusted_ibcmd))} \"$@\"\n" + ).encode("utf-8") + _write_new_file(wrapper, payload, mode=0o700) + return wrapper_platform, { + "builder": "IBCMD", + "privateIbcmdData": True, + "buildPlatformPath": str(wrapper_platform), + "buildDataPath": str(private_data), + "fullDumpPlatformPath": str(trusted_platform), + "wrapperSha256": hashlib.sha256(payload).hexdigest(), + "trustedIbcmdSha256": _hash_file(trusted_ibcmd), + } + + +def _redaction_pairs(redactions) -> list[tuple[str, str, bool]]: + pairs: set[tuple[str, str, bool]] = set() + for raw, replacement in redactions or []: + value = str(raw) + if not value: + continue + path_like = isinstance(raw, os.PathLike) + pairs.add((value, str(replacement), path_like)) + if not path_like: + continue + try: + resolved = str(Path(raw).resolve()) + except (OSError, TypeError, ValueError): + continue + if resolved: + pairs.add((resolved, str(replacement), True)) + return sorted(pairs, key=lambda item: -len(item[0])) + + +def _sanitize_text( + value: str, + redactions, + *, + limit: int | None = None, + redact_tokens: bool = True, +) -> str: + text = value + for raw, replacement, path_like in _redaction_pairs(redactions): + if path_like: + text = text.replace(raw, replacement) + elif redact_tokens: + if any(character.isalnum() for character in raw): + token = re.compile(rf"(?", text) + text = _CONNECTION_CREDENTIAL_RE.sub("", text) + text = _CLI_CREDENTIAL_RE.sub("", text) + if limit is not None and len(text) > limit: + text = text[:limit] + "…" + return text + + +def _sanitize_key(value: str, redactions) -> str: + text = value + for raw, replacement, path_like in _redaction_pairs(redactions): + if path_like: + text = text.replace(raw, replacement) + return text + + +def _is_secret_key(value: str) -> bool: + normalized = value.casefold().lstrip("-") + return normalized in _EXACT_SECRET_KEYS or any( + marker in normalized for marker in _SUBSTRING_SECRET_KEYS + ) + + +def _sanitize_value(value, redactions, *, redact_tokens: bool = True): + if isinstance(value, str): + return _sanitize_text(value, redactions, redact_tokens=redact_tokens) + if isinstance(value, Path): + return _sanitize_text( + str(value), + redactions, + redact_tokens=redact_tokens, + ) + if isinstance(value, dict): + sanitized = {} + for key, child in value.items(): + key_text = str(key) + sanitized_key = _sanitize_key(key_text, redactions) + sanitized[sanitized_key] = ( + "" + if _is_secret_key(key_text) + else _sanitize_value( + child, + redactions, + redact_tokens=redact_tokens, + ) + ) + return sanitized + if isinstance(value, list): + return [ + _sanitize_value(child, redactions, redact_tokens=redact_tokens) + for child in value + ] + if isinstance(value, tuple): + return [ + _sanitize_value(child, redactions, redact_tokens=redact_tokens) + for child in value + ] + return value + + +def _filtered_child_environment(source) -> tuple[dict[str, str], list[tuple[str, str]]]: + environment: dict[str, str] = {} + secret_redactions: list[tuple[str, str]] = [] + for raw_name, raw_value in source.items(): + name = str(raw_name) + value = str(raw_value) + upper_name = name.upper() + normalized = re.sub(r"[^A-Z0-9]+", "_", upper_name) + name_parts = frozenset(part for part in normalized.split("_") if part) + sensitive = ( + upper_name in _SENSITIVE_ENVIRONMENT_NAMES + or bool(name_parts.intersection({"JWT", "PAT"})) + or any( + marker in normalized for marker in _SENSITIVE_ENVIRONMENT_MARKERS + ) + ) + if sensitive: + if value: + secret_redactions.append((value, "$ENV_SECRET")) + continue + allowed = upper_name in _CHILD_ENVIRONMENT_NAMES or any( + upper_name.startswith(prefix) for prefix in _CHILD_ENVIRONMENT_PREFIXES + ) + if allowed: + environment[name] = value + return environment, secret_redactions + + +def _json_digest(value) -> str: + payload = json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + default=str, + ).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +def _snapshot_tree(root: Path) -> dict: + if root.is_symlink() or not root.is_dir(): + raise SourceError(f"source snapshot root is not a safe directory: {root}") + files: dict[str, str] = {} + empty_directories: list[str] = [] + identities: set[tuple[int, int]] = set() + + def visit(directory: Path) -> None: + try: + with os.scandir(directory) as iterator: + entries = sorted(iterator, key=lambda entry: entry.name) + except OSError as error: + raise SourceError(f"cannot enumerate source snapshot {directory}: {error}") from error + if not entries and directory != root: + empty_directories.append(directory.relative_to(root).as_posix()) + for entry in entries: + path = Path(entry.path) + if entry.is_symlink(): + raise SourceError(f"source snapshot symlink is forbidden: {path}") + try: + metadata = entry.stat(follow_symlinks=False) + except OSError as error: + raise SourceError(f"cannot inspect source snapshot {path}: {error}") from error + identity = (metadata.st_dev, metadata.st_ino) + if identity in identities: + raise SourceError(f"source snapshot hardlink alias is forbidden: {path}") + identities.add(identity) + if entry.is_dir(follow_symlinks=False): + visit(path) + elif entry.is_file(follow_symlinks=False): + files[path.relative_to(root).as_posix()] = _hash_file(path) + else: + raise SourceError(f"source snapshot special entry is forbidden: {path}") + + visit(root) + return {"files": files, "emptyDirectories": sorted(empty_directories)} + + +def _snapshot_digest(snapshot: dict) -> str: + return _json_digest(snapshot) + + +def _optional_file_hash(path: Path) -> str | None: + if not path.exists(): + return None + _require_regular_file(path, "optional evidence file") + return _hash_file(path) + + +def _restore_private_preimage( + path: Path, + payload: bytes, + *, + expected_current_sha256: str, +) -> None: + """Atomically restore a private target after proving its mutation preimage.""" + + _require_unlinked_directory_ancestors(path, "round-trip source target") + _require_regular_file(path, "round-trip source target") + if _hash_file(path) != expected_current_sha256: + raise SourceError( + "round-trip source target changed before the pre-dump oracle reset: " + f"{path}" + ) + temporary_path = None + try: + with tempfile.NamedTemporaryFile( + mode="wb", + dir=path.parent, + prefix=f".{path.name}.issue-76-reset.", + delete=False, + ) as stream: + temporary_path = Path(stream.name) + os.chmod(temporary_path, 0o600) + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + _require_unlinked_directory_ancestors(path, "round-trip source target") + _require_regular_file(path, "round-trip source target") + if _hash_file(path) != expected_current_sha256: + raise SourceError( + "round-trip source target changed while the pre-dump oracle reset " + f"was prepared: {path}" + ) + os.replace(temporary_path, path) + except Exception as error: + if temporary_path is not None: + try: + _require_unlinked_directory_ancestors( + temporary_path, + "round-trip temporary target", + ) + temporary_path.unlink(missing_ok=True) + except (OSError, SourceError): + pass + if isinstance(error, SourceError): + raise + raise SourceError(f"cannot restore private round-trip preimage: {error}") from error + + +def _metadata_marker_survived(path: Path, marker: str) -> bool: + try: + root = ElementTree.fromstring(path.read_bytes()) + except (OSError, ElementTree.ParseError): + return False + comments = [ + element + for element in root.iter() + if element.tag.rsplit("}", 1)[-1] == "Comment" + ] + return len(comments) == 1 and (comments[0].text or "") == marker + + +def _metadata_marker_present(path: Path, marker: str) -> bool: + try: + root = ElementTree.fromstring(path.read_bytes()) + except (OSError, ElementTree.ParseError): + return False + return any( + element.tag.rsplit("}", 1)[-1] == "Comment" + and (element.text or "") == marker + for element in root.iter() + ) + + +def _metadata_issue_marker_present(path: Path) -> bool: + try: + root = ElementTree.fromstring(path.read_bytes()) + except (OSError, ElementTree.ParseError): + return False + return any( + element.tag.rsplit("}", 1)[-1] == "Comment" + and (element.text or "").startswith(MARKER_PREFIX) + for element in root.iter() + ) + + +def _module_marker_survived(path: Path, marker: str) -> bool: + try: + text = path.read_bytes().decode("utf-8-sig") + except (OSError, UnicodeDecodeError): + return False + marker_index = text.find(marker) + method_index = text.find("ЛинияПоддержки") + return ( + marker_index >= 0 + and text.count(marker) == 1 + and method_index > marker_index + ) + + +def _module_marker_present(path: Path, marker: str) -> bool: + try: + return marker in path.read_bytes().decode("utf-8-sig") + except (OSError, UnicodeDecodeError): + return False + + +def _new_scenario_marker() -> str: + return f"{MARKER_PREFIX}_{secrets.token_hex(16).upper()}" + + +def _scenario_markers(marker: str | None) -> tuple[str, str]: + selected = _new_scenario_marker() if marker is None else marker + if not isinstance(selected, str) or MARKER_RE.fullmatch(selected) is None: + raise SourceError("scenario marker has an invalid format") + return selected, f"// {selected}" + + +def _text_profile(path: Path) -> dict | None: + try: + payload = path.read_bytes() + except OSError: + return None + crlf = payload.count(b"\r\n") + bare_lf = payload.count(b"\n") - crlf + return { + "bomPrefixBytes": 3 if payload.startswith(b"\xef\xbb\xbf") else 0, + "crlfCount": crlf, + "bareLfCount": bare_lf, + "terminalNewline": payload.endswith((b"\r", b"\n")), + } + + +def _step_record( + *, + step_id: str, + tool: str, + arguments: dict, + payload: dict, + duration_ms: int, + redactions, +) -> dict: + sanitized_arguments = _sanitize_value( + arguments, + redactions, + redact_tokens=False, + ) + sanitized_payload = _sanitize_value(payload, redactions) + projection = { + "ok": payload.get("ok"), + "summary": _sanitize_text( + str(payload.get("summary", "")), redactions, limit=DIAGNOSTIC_LIMIT + ), + "warnings": [ + _sanitize_text(str(item), redactions, limit=DIAGNOSTIC_LIMIT) + for item in payload.get("warnings", []) + ], + "errors": [ + _sanitize_text(str(item), redactions, limit=DIAGNOSTIC_LIMIT) + for item in payload.get("errors", []) + ], + } + for stream in ("stdout", "stderr"): + if payload.get(stream) not in (None, ""): + projection[stream] = _sanitize_text( + str(payload[stream]), redactions, limit=DIAGNOSTIC_LIMIT + ) + return { + "id": step_id, + "tool": tool, + "arguments": sanitized_arguments, + "argumentsSha256": _json_digest(sanitized_arguments), + "resultSha256": _json_digest(sanitized_payload), + "result": projection, + "durationMs": duration_ms, + } + + +def _invoke_step(client, report: dict, step_id: str, tool: str, arguments: dict, redactions): + started = time.monotonic() + try: + payload = client.call(tool, arguments) + except SourceError: + raise + except Exception as error: + raise SourceError(f"{tool} call failed at {step_id}: {error}") from error + duration_ms = int(round((time.monotonic() - started) * 1000)) + if not isinstance(payload, dict): + raise SourceError(f"{tool} returned a non-object payload at {step_id}") + report["steps"].append( + _step_record( + step_id=step_id, + tool=tool, + arguments=arguments, + payload=payload, + duration_ms=duration_ms, + redactions=redactions, + ) + ) + return payload + + +def _new_flow_report( + workspace: Path, + *, + metadata_marker: str, + bsl_marker: str, +) -> dict: + catalog = workspace / CATALOG_RELATIVE_PATH + module = workspace / MODULE_RELATIVE_PATH + return { + "schemaVersion": 1, + "scenario": SCENARIO, + "status": "failed", + "exitCode": 1, + "steps": [], + "builds": { + "baselineBuild": {"stepId": "baseline-build", "ok": False}, + "mutationBuild": {"stepId": "mutation-build", "ok": False}, + }, + "partialGuard": { + "blocked": False, + "sourceUnchanged": False, + "beforeSha256": None, + "afterSha256": None, + }, + "roundTrip": { + "metadata": { + "metadataPath": CATALOG_METADATA_PATH, + "marker": metadata_marker, + "beforeSha256": _optional_file_hash(catalog), + "afterMutationSha256": None, + "beforeFullDumpSha256": None, + "presentAfterMutation": False, + "absentBeforeFullDump": False, + "afterFullDumpSha256": None, + "survived": False, + }, + "module": { + "metadataPath": MODULE_METADATA_PATH, + "marker": bsl_marker, + "beforeSha256": _optional_file_hash(module), + "afterMutationSha256": None, + "beforeFullDumpSha256": None, + "presentAfterMutation": False, + "absentBeforeFullDump": False, + "afterFullDumpSha256": None, + "survived": False, + "textProfile": None, + }, + }, + "configDumpInfo": { + "before": _optional_file_hash(workspace / CONFIG_DUMP_INFO_RELATIVE_PATH), + "afterBaselineBuild": None, + "afterBuild": None, + "afterFullDump": None, + "changedByBaselineBuild": None, + "changedByBuild": None, + "changedByFullDump": None, + "informationalOnly": True, + }, + "summary": {"failures": []}, + } + + +def _finish_flow(report: dict, status: str, exit_code: int, redactions, failure=None): + if failure: + report["summary"]["failures"].append(str(failure)) + report["status"] = status + report["exitCode"] = exit_code + report["summary"]["stepCount"] = len(report["steps"]) + report["summary"]["passed"] = status == "pass" + return exit_code, _sanitize_value(report, redactions, redact_tokens=False) + + +def _payload_ok(payload: dict) -> bool: + return payload.get("ok") is True + + +def _support_state_matches(payload: dict, *, editing_enabled: bool) -> bool: + data = payload.get("data") + support = data.get("support") if isinstance(data, dict) else None + return ( + isinstance(support, dict) + and support.get("state") == "supported" + and support.get("editingEnabled") is editing_enabled + ) + + +def _support_apply_matches(payload: dict, *, action: str) -> bool: + data = payload.get("data") + if not isinstance(data, dict): + return False + if data.get("action") != action or data.get("applied") is not True: + return False + if action == "capability": + return data.get("editingEnabled") is True + records_changed = data.get("recordsChanged") + return ( + action == "objectRule" + and isinstance(records_changed, int) + and not isinstance(records_changed, bool) + and records_changed > 0 + and data.get("rule") in EDITABLE_SUPPORT_RULE_RECEIPTS + ) + + +def _guard_is_blocked(payload: dict) -> bool: + if payload.get("ok") is not False: + return False + combined = "\n".join( + [str(payload.get("summary", ""))] + + [str(item) for item in payload.get("errors", [])] + ).casefold() + return ( + "source sync guard" in combined + and "v8-runner-rust#30" in combined + and "divergence-safe merge" in combined + ) + + +def run_roundtrip_flow( + client, + *, + workspace: Path, + redactions=None, + marker: str | None = None, + before_full_dump=None, +) -> tuple[int, dict]: + """Run issue #76 against one already-private workspace. + + ``client`` is the intentionally small test seam: it provides + ``call(tool_name, arguments) -> dict``. Production supplies the MCP stdio + client below; unit tests supply a scripted public-tool fake. + """ + + workspace = _resolved_absolute( + Path(workspace), + "private round-trip workspace", + directory=True, + ) + source = workspace / "src" + catalog = workspace / CATALOG_RELATIVE_PATH + common_module_descriptor = workspace / COMMON_MODULE_DESCRIPTOR_RELATIVE_PATH + module = workspace / MODULE_RELATIVE_PATH + for path, label in ( + (source / "Configuration.xml", "configuration descriptor"), + (catalog, "catalog descriptor"), + (common_module_descriptor, "common module descriptor"), + (module, "common module source"), + ): + _require_unlinked_directory_ancestors(path, label) + _require_regular_file(path, label) + + redactions = list(redactions or []) + metadata_marker, bsl_marker = _scenario_markers(marker) + report = _new_flow_report( + workspace, + metadata_marker=metadata_marker, + bsl_marker=bsl_marker, + ) + try: + catalog_preimage = catalog.read_bytes() + module_preimage = module.read_bytes() + except OSError as error: + raise SourceError(f"cannot read round-trip source preimages: {error}") from error + + preexisting_markers = [] + if _metadata_issue_marker_present(catalog): + preexisting_markers.append("metadata") + if _module_marker_present(module, f"// {MARKER_PREFIX}"): + preexisting_markers.append("module") + if preexisting_markers: + return _finish_flow( + report, + "failed", + 1, + redactions, + "scenario marker already present before baseline build: " + + ", ".join(preexisting_markers), + ) + + def invoke(step_id: str, tool: str, arguments: dict) -> dict: + return _invoke_step(client, report, step_id, tool, arguments, redactions) + + cf_before = invoke( + "support-info-before", + "unica.cf.info", + {"sourceSet": SOURCE_SET}, + ) + if not _payload_ok(cf_before): + return _finish_flow(report, "failed", 1, redactions, "initial cf.info failed") + if not _support_state_matches(cf_before, editing_enabled=False): + return _finish_flow( + report, + "failed", + 1, + redactions, + "support precondition is not a supported configuration with editing disabled", + ) + + support_operations = [ + ( + "support-capability", + {"Path": "src", "Capability": "on"}, + ), + ( + "support-catalog", + {"Path": CATALOG_RELATIVE_PATH.as_posix(), "Set": "editable"}, + ), + ( + "support-common-module", + { + "Path": COMMON_MODULE_DESCRIPTOR_RELATIVE_PATH.as_posix(), + "Set": "editable", + }, + ), + ] + for step_id, operation in support_operations: + for dry_run, suffix in ((True, "preview"), (False, "apply")): + arguments = { + "cwd": str(workspace), + **operation, + "dryRun": dry_run, + } + payload = invoke( + f"{step_id}-{suffix}", + "unica.support.edit", + arguments, + ) + if not _payload_ok(payload): + return _finish_flow( + report, + "failed", + 1, + redactions, + f"{step_id} {suffix} failed", + ) + if not dry_run: + expected_action = ( + "capability" if step_id == "support-capability" else "objectRule" + ) + if not _support_apply_matches(payload, action=expected_action): + return _finish_flow( + report, + "failed", + 1, + redactions, + f"{step_id} apply did not report the required support transition", + ) + + cf_after = invoke( + "support-info-after", + "unica.cf.info", + {"sourceSet": SOURCE_SET}, + ) + if not _payload_ok(cf_after): + return _finish_flow(report, "failed", 1, redactions, "final cf.info failed") + if not _support_state_matches(cf_after, editing_enabled=True): + return _finish_flow( + report, + "failed", + 1, + redactions, + "support postcondition did not confirm enabled configuration editing", + ) + + baseline_build = invoke( + "baseline-build", + "unica.runtime.execute", + { + "cwd": str(workspace), + "operation": "build", + "sourceSet": SOURCE_SET, + "fullRebuild": True, + "dryRun": False, + }, + ) + report["builds"]["baselineBuild"]["ok"] = _payload_ok(baseline_build) + if not _payload_ok(baseline_build): + return _finish_flow( + report, + "failed", + 1, + redactions, + "support baseline build failed", + ) + cdfi = workspace / CONFIG_DUMP_INFO_RELATIVE_PATH + report["configDumpInfo"]["afterBaselineBuild"] = _optional_file_hash(cdfi) + report["configDumpInfo"]["changedByBaselineBuild"] = ( + report["configDumpInfo"]["afterBaselineBuild"] + != report["configDumpInfo"]["before"] + ) + + meta_arguments = { + "sourceSet": SOURCE_SET, + "metadataPath": CATALOG_METADATA_PATH, + "operations": [ + { + "op": "setProperties", + "values": {"Comment": metadata_marker}, + } + ], + } + for dry_run, suffix in ((True, "preview"), (False, "apply")): + payload = invoke( + f"meta-edit-{suffix}", + "unica.meta.edit", + {**meta_arguments, "dryRun": dry_run}, + ) + if not _payload_ok(payload): + return _finish_flow( + report, "failed", 1, redactions, f"meta.edit {suffix} failed" + ) + + code_arguments = { + "cwd": str(workspace), + "sourceSet": SOURCE_SET, + "metadataPath": MODULE_METADATA_PATH, + "operation": "insert", + "selector": {"method": "ЛинияПоддержки"}, + "position": "before", + "content": bsl_marker, + } + for dry_run, suffix in ((True, "preview"), (False, "apply")): + payload = invoke( + f"code-patch-{suffix}", + "unica.code.patch", + {**code_arguments, "dryRun": dry_run}, + ) + if not _payload_ok(payload): + return _finish_flow( + report, "failed", 1, redactions, f"code.patch {suffix} failed" + ) + + report["roundTrip"]["metadata"]["afterMutationSha256"] = _hash_file(catalog) + report["roundTrip"]["module"]["afterMutationSha256"] = _hash_file(module) + report["roundTrip"]["metadata"]["presentAfterMutation"] = ( + _metadata_marker_survived(catalog, metadata_marker) + ) + report["roundTrip"]["module"]["presentAfterMutation"] = ( + _module_marker_survived(module, bsl_marker) + ) + if not all( + ( + report["roundTrip"]["metadata"]["presentAfterMutation"], + report["roundTrip"]["module"]["presentAfterMutation"], + ) + ): + return _finish_flow( + report, + "failed", + 1, + redactions, + "one or both source mutations were not observable before runtime", + ) + + before_partial = _snapshot_tree(source) + partial = invoke( + "partial-dump-guard", + "unica.runtime.execute", + { + "cwd": str(workspace), + "operation": "dump", + "mode": "partial", + "object": "Catalog:ЗависимостиСчетов", + "sourceSet": SOURCE_SET, + "dryRun": False, + }, + ) + after_partial = _snapshot_tree(source) + report["partialGuard"].update( + { + "blocked": _guard_is_blocked(partial), + "sourceUnchanged": before_partial == after_partial, + "beforeSha256": _snapshot_digest(before_partial), + "afterSha256": _snapshot_digest(after_partial), + } + ) + if not report["partialGuard"]["blocked"]: + return _finish_flow( + report, + "failed", + 1, + redactions, + "applied partial dump was not rejected by the issue #76 source sync guard", + ) + if not report["partialGuard"]["sourceUnchanged"]: + return _finish_flow( + report, + "failed", + 1, + redactions, + "the rejected applied partial dump changed the private source tree", + ) + + build = invoke( + "mutation-build", + "unica.runtime.execute", + { + "cwd": str(workspace), + "operation": "build", + "sourceSet": SOURCE_SET, + "dryRun": False, + }, + ) + report["builds"]["mutationBuild"]["ok"] = _payload_ok(build) + if not _payload_ok(build): + return _finish_flow(report, "failed", 1, redactions, "mutation build failed") + + report["configDumpInfo"]["afterBuild"] = _optional_file_hash(cdfi) + report["configDumpInfo"]["changedByBuild"] = ( + report["configDumpInfo"]["afterBuild"] + != report["configDumpInfo"]["afterBaselineBuild"] + ) + + if before_full_dump is not None: + before_full_dump() + + _restore_private_preimage( + catalog, + catalog_preimage, + expected_current_sha256=report["roundTrip"]["metadata"][ + "afterMutationSha256" + ], + ) + _restore_private_preimage( + module, + module_preimage, + expected_current_sha256=report["roundTrip"]["module"][ + "afterMutationSha256" + ], + ) + report["roundTrip"]["metadata"]["beforeFullDumpSha256"] = _hash_file(catalog) + report["roundTrip"]["module"]["beforeFullDumpSha256"] = _hash_file(module) + report["roundTrip"]["metadata"]["absentBeforeFullDump"] = not ( + _metadata_marker_present(catalog, metadata_marker) + ) + report["roundTrip"]["module"]["absentBeforeFullDump"] = not ( + _module_marker_present(module, bsl_marker) + ) + if not all( + ( + report["roundTrip"]["metadata"]["absentBeforeFullDump"], + report["roundTrip"]["module"]["absentBeforeFullDump"], + report["roundTrip"]["metadata"]["beforeFullDumpSha256"] + == report["roundTrip"]["metadata"]["beforeSha256"], + report["roundTrip"]["module"]["beforeFullDumpSha256"] + == report["roundTrip"]["module"]["beforeSha256"], + ) + ): + return _finish_flow( + report, + "failed", + 1, + redactions, + "could not establish marker-free source preimages before the full dump", + ) + + full_dump = invoke( + "safe-full-dump", + "unica.runtime.execute", + { + "cwd": str(workspace), + "operation": "dump", + "mode": "full", + "sourceSet": SOURCE_SET, + "dryRun": False, + }, + ) + if not _payload_ok(full_dump): + return _finish_flow(report, "failed", 1, redactions, "safe full dump failed") + + report["configDumpInfo"]["afterFullDump"] = _optional_file_hash(cdfi) + report["configDumpInfo"]["changedByFullDump"] = ( + report["configDumpInfo"]["afterFullDump"] + != report["configDumpInfo"]["afterBuild"] + ) + report["roundTrip"]["metadata"]["afterFullDumpSha256"] = _optional_file_hash( + catalog + ) + report["roundTrip"]["module"]["afterFullDumpSha256"] = _optional_file_hash(module) + report["roundTrip"]["metadata"]["survived"] = _metadata_marker_survived( + catalog, + metadata_marker, + ) + report["roundTrip"]["module"]["survived"] = _module_marker_survived( + module, + bsl_marker, + ) + report["roundTrip"]["module"]["textProfile"] = _text_profile(module) + + lost = [ + name + for name in ("metadata", "module") + if not report["roundTrip"][name]["survived"] + ] + if lost: + return _finish_flow( + report, + "failed", + 1, + redactions, + "marker lost after safe full dump: " + ", ".join(lost), + ) + return _finish_flow(report, "pass", 0, redactions) + + +class McpSession: + """Sequential JSONL MCP client with bounded stderr and request deadlines.""" + + def __init__( + self, + command: list[str], + environment: dict[str, str], + timeout_seconds: float, + *, + cwd: Path, + ) -> None: + if not command or any(not isinstance(item, str) or not item for item in command): + raise SourceError("Unica command must be a non-empty argument array") + self.timeout_seconds = timeout_seconds + self.lines: queue.Queue[str | None] = queue.Queue() + self.diagnostics: deque[str] = deque(maxlen=256) + self.next_id = 1 + popen_options = {} + if os.name == "posix": + popen_options["start_new_session"] = True + elif os.name == "nt" and hasattr(subprocess, "CREATE_NEW_PROCESS_GROUP"): + popen_options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP + try: + self.process = subprocess.Popen( + command, + cwd=cwd, + env=environment, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + **popen_options, + ) + except (OSError, ValueError) as error: + raise SourceError(f"cannot start packaged Unica MCP: {error}") from error + self.stdout_reader = threading.Thread(target=self._read_stdout, daemon=True) + self.stderr_reader = threading.Thread(target=self._read_stderr, daemon=True) + self.stdout_reader.start() + self.stderr_reader.start() + + def _read_stdout(self) -> None: + assert self.process.stdout is not None + try: + for line in self.process.stdout: + self.lines.put(line) + finally: + self.lines.put(None) + + def _read_stderr(self) -> None: + assert self.process.stderr is not None + for line in self.process.stderr: + self.diagnostics.append(line[-DIAGNOSTIC_LIMIT:]) + + def _diagnostic_text(self) -> str: + return "".join(self.diagnostics)[-DIAGNOSTIC_LIMIT:].strip() or "no process output" + + def _terminate(self) -> None: + if self.process.poll() is not None: + return + try: + if os.name == "posix": + os.killpg(self.process.pid, signal.SIGTERM) + else: + self.process.terminate() + except (OSError, ProcessLookupError): + pass + try: + self.process.wait(timeout=2) + except subprocess.TimeoutExpired: + try: + if os.name == "posix": + os.killpg(self.process.pid, signal.SIGKILL) + else: + self.process.kill() + except (OSError, ProcessLookupError): + pass + self.process.wait(timeout=2) + + def request(self, message: dict) -> dict: + if self.process.poll() is not None: + raise SourceError( + f"packaged Unica exited before request: {self._diagnostic_text()}" + ) + assert self.process.stdin is not None + try: + self.process.stdin.write( + json.dumps(message, ensure_ascii=False, separators=(",", ":")) + "\n" + ) + self.process.stdin.flush() + except (OSError, BrokenPipeError) as error: + self._terminate() + raise SourceError(f"cannot write packaged Unica MCP request: {error}") from error + deadline = time.monotonic() + self.timeout_seconds + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + self._terminate() + raise SourceError( + f"packaged Unica MCP request timed out after " + f"{self.timeout_seconds:g}s: {self._diagnostic_text()}" + ) + try: + line = self.lines.get(timeout=remaining) + except queue.Empty as error: + self._terminate() + raise SourceError( + f"packaged Unica MCP request timed out after " + f"{self.timeout_seconds:g}s: {self._diagnostic_text()}" + ) from error + if line is None: + raise SourceError( + f"packaged Unica exited before the expected MCP response: " + f"{self._diagnostic_text()}" + ) + try: + response = json.loads(line) + except json.JSONDecodeError as error: + self._terminate() + raise SourceError(f"packaged Unica emitted invalid JSON: {error}") from error + if isinstance(response, dict) and response.get("id") == message.get("id"): + return response + + def notify(self, message: dict) -> None: + assert self.process.stdin is not None + try: + self.process.stdin.write( + json.dumps(message, ensure_ascii=False, separators=(",", ":")) + "\n" + ) + self.process.stdin.flush() + except (OSError, BrokenPipeError) as error: + self._terminate() + raise SourceError(f"cannot write packaged Unica MCP notification: {error}") from error + + def start(self, required_tools=REQUIRED_TOOLS) -> None: + initialize = self.request( + { + "jsonrpc": "2.0", + "id": self.next_id, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": { + "name": "unica-issue-76-roundtrip", + "version": "1", + }, + }, + } + ) + self.next_id += 1 + if "result" not in initialize: + raise SourceError(f"packaged Unica initialize failed: {initialize.get('error')}") + self.notify( + { + "jsonrpc": "2.0", + "method": "notifications/initialized", + "params": {}, + } + ) + listed = self.request( + { + "jsonrpc": "2.0", + "id": self.next_id, + "method": "tools/list", + "params": {}, + } + ) + self.next_id += 1 + try: + names = { + item["name"] + for item in listed["result"]["tools"] + if isinstance(item, dict) and isinstance(item.get("name"), str) + } + except (KeyError, TypeError) as error: + raise SourceError("packaged Unica tools/list response is malformed") from error + missing = sorted(set(required_tools) - names) + if missing: + raise SourceError("packaged Unica is missing required tools: " + ", ".join(missing)) + + def call(self, name: str, arguments: dict, **_kwargs) -> dict: + response = self.request( + { + "jsonrpc": "2.0", + "id": self.next_id, + "method": "tools/call", + "params": {"name": name, "arguments": arguments}, + } + ) + self.next_id += 1 + if "error" in response: + error = response["error"] + raise SourceError(f"{name} failed as JSON-RPC: {error.get('message', error)}") + try: + result = response["result"] + except (KeyError, TypeError) as error: + raise SourceError(f"{name} response has no MCP result") from error + structured = result.get("structuredContent") if isinstance(result, dict) else None + text_payload = None + try: + content_text = result["content"][0]["text"] + text_payload = json.loads(content_text) + except (KeyError, IndexError, TypeError, json.JSONDecodeError): + text_payload = None + if isinstance(structured, dict): + if isinstance(text_payload, dict) and text_payload != structured: + raise SourceError(f"{name} text and structuredContent results diverged") + return structured + if isinstance(text_payload, dict): + return text_payload + raise SourceError(f"{name} response has no JSON object payload") + + def close(self) -> None: + if self.process.stdin is not None and not self.process.stdin.closed: + try: + self.process.stdin.close() + except OSError: + pass + try: + return_code = self.process.wait(timeout=min(self.timeout_seconds, 10.0)) + except subprocess.TimeoutExpired: + self._terminate() + return_code = self.process.returncode + self.stdout_reader.join(timeout=1) + self.stderr_reader.join(timeout=1) + for stream in (self.process.stdout, self.process.stderr): + if stream is not None and not stream.closed: + stream.close() + if return_code not in (0, None): + raise SourceError( + f"packaged Unica MCP exited with {return_code}: {self._diagnostic_text()}" + ) + + +def _atomic_write_report(path: Path, report: dict) -> None: + payload = ( + json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n" + ).encode("utf-8") + temporary_path = None + try: + with tempfile.NamedTemporaryFile( + mode="wb", + dir=path.parent, + prefix=f".{path.name}.", + delete=False, + ) as stream: + temporary_path = Path(stream.name) + os.chmod(temporary_path, 0o600) + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary_path, path) + except OSError as error: + if temporary_path is not None: + try: + temporary_path.unlink(missing_ok=True) + except OSError: + pass + raise SourceError(f"cannot atomically write issue #76 report: {error}") from error + + +def _source_error_report(error: Exception, redactions) -> dict: + sanitized_message = _sanitize_text(str(error), redactions) + return _sanitize_value( + { + "schemaVersion": 1, + "scenario": SCENARIO, + "status": "source-error", + "exitCode": 2, + "steps": [], + "partialGuard": {"blocked": False, "sourceUnchanged": False}, + "roundTrip": { + "metadata": {"survived": False}, + "module": {"survived": False}, + }, + "configDumpInfo": {"informationalOnly": True}, + "summary": {"passed": False, "stepCount": 0}, + "sourceError": {"message": sanitized_message}, + }, + redactions, + redact_tokens=False, + ) + + +def _packaged_manifest_provenance( + plugin_root: Path, + *, + unica_binary_sha256: str, +) -> dict: + if _SHA256_RE.fullmatch(unica_binary_sha256) is None: + raise SourceError("executed Unica binary has an invalid sha256") + path = _require_regular_file( + plugin_root / "third-party/manifest.json", + "packaged plugin tool manifest", + ) + try: + manifest_bytes = path.read_bytes() + payload = json.loads(manifest_bytes) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error: + raise SourceError(f"cannot read packaged plugin tool manifest: {error}") from error + if not isinstance(payload, dict): + raise SourceError("packaged plugin tool manifest must be a JSON object") + schema_version = payload.get("schemaVersion") + target_triple = payload.get("targetTriple") + tools = payload.get("tools") + if isinstance(schema_version, bool) or not isinstance(schema_version, int): + raise SourceError("packaged plugin tool manifest has invalid schemaVersion") + if not isinstance(target_triple, str) or not target_triple: + raise SourceError("packaged plugin tool manifest has invalid targetTriple") + if not isinstance(tools, list): + raise SourceError("packaged plugin tool manifest has invalid tools array") + + selected = {} + for item in tools: + if not isinstance(item, dict) or item.get("name") not in _MANIFEST_TOOL_NAMES: + continue + name = item["name"] + if name in selected: + raise SourceError(f"packaged plugin tool manifest duplicates {name}") + projection = {} + for field in _MANIFEST_TOOL_FIELDS: + value = item.get(field) + if not isinstance(value, str) or not value: + raise SourceError( + f"packaged plugin tool manifest {name} has invalid {field}" + ) + projection[field] = value + if _SHA256_RE.fullmatch(projection["sha256"]) is None: + raise SourceError( + f"packaged plugin tool manifest {name} has invalid sha256" + ) + selected[name] = projection + missing = sorted(set(_MANIFEST_TOOL_NAMES) - set(selected)) + if missing: + raise SourceError( + "packaged plugin tool manifest is missing required tools: " + + ", ".join(missing) + ) + if selected["unica"]["sha256"] != unica_binary_sha256: + raise SourceError( + "packaged plugin manifest Unica sha256 does not match the executed binary" + ) + return { + "sha256": hashlib.sha256(manifest_bytes).hexdigest(), + "schemaVersion": schema_version, + "targetTriple": target_triple, + "tools": selected, + } + + +def _record_source_error(report: dict, message: str) -> None: + report["status"] = "source-error" + report["exitCode"] = 2 + summary = report.setdefault("summary", {}) + summary["passed"] = False + failures = summary.setdefault("failures", []) + if message not in failures: + failures.append(message) + report["sourceError"] = {"message": message} + + +def execute_gate( + *, + binary: Path, + binary_args: list[str], + plugin_root: Path, + database: Path, + sources: Path, + parent_configuration: Path, + platform_path: Path, + platform_version: str, + report_path: Path, + evidence_dir: Path | None, + builder: str, + db_user: str, + timeout_seconds: float, + execute: bool, + allow_empty_password: bool, + session_factory=McpSession, +) -> tuple[int, dict]: + if execute is not True or allow_empty_password is not True: + raise SourceError("both live mutation opt-ins must be explicit") + if builder not in {"DESIGNER", "IBCMD"}: + raise SourceError("builder must be DESIGNER or IBCMD") + if PLATFORM_VERSION_RE.fullmatch(platform_version) is None: + raise SourceError("platform version must be exact 8.3.27.x") + if ( + not math.isfinite(timeout_seconds) + or timeout_seconds <= 0 + or timeout_seconds > MAX_TIMEOUT_SECONDS + ): + raise SourceError("timeout must be finite, positive, and no more than 24 hours") + + initial_redactions = [ + (database, "$DATABASE_INPUT"), + (sources, "$SOURCE_INPUT"), + (parent_configuration, "$PARENT_CONFIGURATION_INPUT"), + (platform_path, "$PLATFORM"), + (plugin_root, "$PLUGIN_ROOT"), + (binary, "$UNICA_BINARY"), + (db_user, "$DB_USER"), + ] + database_root = _resolved_absolute(database, "database input", directory=True) + sources_root = _resolved_absolute(sources, "source input", directory=True) + parent_configuration_path = _resolved_absolute( + parent_configuration, + "parent configuration input", + directory=False, + ) + binary_path = _resolved_absolute(binary, "Unica binary", directory=False) + plugin = _resolved_absolute(plugin_root, "plugin root", directory=True) + platform = _resolved_absolute(platform_path, "platform path", directory=True) + if not os.access(binary_path, os.X_OK): + raise SourceError("Unica binary is not executable") + if _paths_overlap(database_root, sources_root): + raise SourceError("database and source inputs must not overlap") + if _is_relative_to(parent_configuration_path, database_root) or _is_relative_to( + parent_configuration_path, + sources_root, + ): + raise SourceError( + "parent configuration input must be outside the database and source trees" + ) + _require_regular_file(database_root / "1Cv8.1CD", "file infobase payload") + for relative, label in ( + (Path("Configuration.xml"), "configuration descriptor"), + (CATALOG_RELATIVE_PATH.relative_to("src"), "catalog descriptor"), + ( + COMMON_MODULE_DESCRIPTOR_RELATIVE_PATH.relative_to("src"), + "common module descriptor", + ), + (MODULE_RELATIVE_PATH.relative_to("src"), "common module source"), + ): + _require_regular_file(sources_root / relative, label) + + protected_mutation_paths = ( + (database_root, "the database input"), + (sources_root, "the source input"), + (parent_configuration_path, "the parent configuration input"), + (binary_path, "the Unica executable"), + (plugin, "the plugin root"), + (platform, "the platform root"), + (Path(__file__).resolve().parents[2], "the repository"), + ) + report_target = _validate_report_path( + report_path, + protected_paths=protected_mutation_paths, + ) + temporary = None + evidence = None + redactions = list(initial_redactions) + report = None + exit_code = 2 + integrity_probe = None + private_binary_started = False + + try: + binary_state_before = _regular_file_stat_signature(binary_path) + binary_sha256_before = _hash_file(binary_path) + plugin_manifest_path = plugin / "third-party/manifest.json" + plugin_manifest = _packaged_manifest_provenance( + plugin, + unica_binary_sha256=binary_sha256_before, + ) + plugin_manifest_state_before = _regular_file_stat_signature( + plugin_manifest_path + ) + if evidence_dir is None: + temporary_parent = _safe_automatic_evidence_parent( + protected_paths=protected_mutation_paths, + ) + try: + temporary = tempfile.TemporaryDirectory( + prefix="unica-issue-76-", + dir=temporary_parent, + ) + except OSError as error: + raise SourceError(f"cannot create private evidence directory: {error}") from error + evidence = _validate_evidence_directory( + Path(temporary.name), + database=database_root, + sources=sources_root, + report_path=report_target, + protected_paths=protected_mutation_paths, + ) + else: + evidence = _validate_evidence_directory( + evidence_dir, + database=database_root, + sources=sources_root, + report_path=report_target, + protected_paths=protected_mutation_paths, + ) + redactions.extend( + [ + (evidence, "$EVIDENCE"), + (evidence / "workspace", "$EVIDENCE/workspace"), + ] + ) + + private_binary = evidence / "unica-executable" + private_binary_sha256, private_binary_size = _copy_regular_file( + binary_path, + private_binary, + ) + if private_binary_sha256 != binary_sha256_before: + raise SourceError("Unica binary changed while making the private execution copy") + private_binary_receipt = { + "sha256": private_binary_sha256, + "bytes": private_binary_size, + "path": "$EVIDENCE/unica-executable", + } + + database_state_before = _stat_tree_digest(database_root) + source_state_before = _stat_tree_digest(sources_root) + parent_state_before = _regular_file_stat_signature(parent_configuration_path) + workspace = evidence / "workspace" + workspace.mkdir(mode=0o700) + database_copy = workspace / "ib" + source_copy = workspace / "src" + database_receipt = _copy_regular_tree(database_root, database_copy) + source_receipt = _copy_regular_tree(sources_root, source_copy) + parent_receipt = _install_parent_configuration( + parent_configuration_path, + source_copy, + ) + private_work = workspace / "work" + private_work.mkdir(mode=0o700) + cache = evidence / "cache" + cache.mkdir(mode=0o700) + runtime_platform = platform + before_full_dump = None + runtime_isolation = { + "builder": builder, + "privateIbcmdData": None, + "buildPlatformPath": str(platform), + "buildDataPath": None, + "fullDumpPlatformPath": str(platform), + } + if builder == "IBCMD": + runtime_platform, runtime_isolation = _create_private_ibcmd_platform( + evidence, + trusted_platform=platform, + platform_version=platform_version, + ) + + def switch_to_trusted_full_dump_platform() -> None: + _replace_project_platform_path( + workspace, + previous_path=runtime_platform, + next_path=platform, + ) + + before_full_dump = switch_to_trusted_full_dump_platform + _write_project_configuration( + workspace, + database_copy=database_copy, + platform_path=runtime_platform, + platform_version=platform_version, + db_user=db_user, + builder=builder, + timeout_seconds=timeout_seconds, + ) + + def inspect_input_integrity() -> dict: + database_state_after = _stat_tree_digest(database_root) + source_state_after = _stat_tree_digest(sources_root) + parent_state_after = _regular_file_stat_signature( + parent_configuration_path + ) + parent_hash_after = _hash_file(parent_configuration_path) + binary_state_after = _regular_file_stat_signature(binary_path) + binary_sha256_after = _hash_file(binary_path) + plugin_manifest_state_after = _regular_file_stat_signature( + plugin_manifest_path + ) + plugin_manifest_sha256_after = _hash_file(plugin_manifest_path) + parent_unchanged = ( + parent_state_before == parent_state_after + and parent_receipt["sha256"] == parent_hash_after + ) + binary_unchanged = ( + binary_state_before == binary_state_after + and binary_sha256_before == binary_sha256_after + ) + plugin_manifest_unchanged = ( + plugin_manifest_state_before == plugin_manifest_state_after + and plugin_manifest["sha256"] == plugin_manifest_sha256_after + ) + database_unchanged = database_state_before == database_state_after + sources_unchanged = source_state_before == source_state_after + unchanged = ( + database_unchanged + and sources_unchanged + and parent_unchanged + and binary_unchanged + and plugin_manifest_unchanged + ) + return { + "unchanged": unchanged, + "binaryUnchanged": binary_unchanged, + "pluginManifestUnchanged": plugin_manifest_unchanged, + "inputs": { + "database": { + "path": "$DATABASE_INPUT", + "copy": database_receipt, + "statUnchanged": database_unchanged, + }, + "sources": { + "path": "$SOURCE_INPUT", + "copy": source_receipt, + "statUnchanged": sources_unchanged, + }, + "parentConfiguration": { + "path": "$PARENT_CONFIGURATION_INPUT", + "copy": parent_receipt, + "statUnchanged": parent_state_before == parent_state_after, + "hashUnchanged": parent_receipt["sha256"] + == parent_hash_after, + }, + "privateCopiesOnly": unchanged, + }, + } + + integrity_probe = inspect_input_integrity + + environment, secret_environment_redactions = _filtered_child_environment( + os.environ + ) + redactions.extend(secret_environment_redactions) + environment["UNICA_PLUGIN_ROOT"] = str(plugin) + environment["UNICA_CACHE_DIR"] = str(cache) + environment["PWD"] = str(workspace) + for temporary_name in ("TMPDIR", "TMP", "TEMP"): + environment[temporary_name] = str(private_work) + command = [str(private_binary), *binary_args] + session = session_factory( + command, + environment, + timeout_seconds, + cwd=workspace, + ) + private_binary_started = True + close_error = None + try: + session.start(REQUIRED_TOOLS) + exit_code, report = run_roundtrip_flow( + session, + workspace=workspace, + redactions=redactions, + before_full_dump=before_full_dump, + ) + finally: + try: + session.close() + except SourceError as error: + close_error = error + except Exception as error: + close_error = SourceError(f"MCP session close failed: {error}") + if close_error is not None: + raise close_error + + integrity = integrity_probe() + report["provenance"] = { + "unicaBinarySha256": binary_sha256_before, + "unicaBinaryCopy": private_binary_receipt, + "executedPrivateBinaryCopy": private_binary_started, + "unicaBinaryUnchanged": integrity["binaryUnchanged"], + "pluginManifest": plugin_manifest, + "pluginManifestUnchanged": integrity["pluginManifestUnchanged"], + "platformVersion": platform_version, + "platformPath": "$PLATFORM", + "builder": builder, + } + report["runtimeIsolation"] = runtime_isolation + report["inputs"] = integrity["inputs"] + report["evidence"] = { + "retained": evidence_dir is not None, + "cleanupSucceeded": None if evidence_dir is not None else False, + "workspace": "$EVIDENCE/workspace", + "containsProprietaryParentConfiguration": True, + } + if not integrity["unchanged"]: + exit_code = 2 + _record_source_error( + report, + "an input tree changed while the private live scenario ran", + ) + report = _sanitize_value(report, redactions, redact_tokens=False) + except Exception as error: + exit_code = 2 + report = _source_error_report(error, redactions) + if integrity_probe is not None: + try: + integrity = integrity_probe() + except Exception as integrity_error: + _record_source_error( + report, + f"{report['sourceError']['message']}; input integrity check failed: {integrity_error}", + ) + else: + report["inputs"] = integrity["inputs"] + report["provenance"] = { + "unicaBinarySha256": binary_sha256_before, + "unicaBinaryCopy": private_binary_receipt, + "executedPrivateBinaryCopy": private_binary_started, + "unicaBinaryUnchanged": integrity["binaryUnchanged"], + "pluginManifest": plugin_manifest, + "pluginManifestUnchanged": integrity[ + "pluginManifestUnchanged" + ], + "platformVersion": platform_version, + "platformPath": "$PLATFORM", + "builder": builder, + } + report["runtimeIsolation"] = runtime_isolation + if not integrity["unchanged"]: + _record_source_error( + report, + f"{report['sourceError']['message']}; an input tree changed while the private live scenario ran", + ) + if evidence is not None: + private_parent = evidence / "workspace" / PARENT_CONFIGURATION_RELATIVE_PATH + report["evidence"] = { + "retained": evidence_dir is not None, + "cleanupSucceeded": None if evidence_dir is not None else False, + "workspace": "$EVIDENCE/workspace", + "containsProprietaryParentConfiguration": private_parent.is_file(), + } + + if temporary is not None: + try: + temporary.cleanup() + except Exception as error: + cleanup_message = f"temporary issue #76 evidence cleanup failed: {error}" + exit_code = 2 + evidence_report = report.setdefault("evidence", {}) + evidence_report.update( + { + "retained": True, + "cleanupSucceeded": False, + "containsProprietaryParentConfiguration": ( + evidence is not None + and ( + evidence + / "workspace" + / PARENT_CONFIGURATION_RELATIVE_PATH + ).is_file() + ), + } + ) + if evidence is not None: + evidence_report["workspace"] = "$EVIDENCE/workspace" + else: + evidence_report.pop("workspace", None) + prior_message = report.get("sourceError", {}).get("message") + terminal_message = ( + f"{prior_message}; {cleanup_message}" + if prior_message + else cleanup_message + ) + _record_source_error(report, terminal_message) + else: + evidence_report = report.setdefault("evidence", {}) + evidence_report["retained"] = False + evidence_report["cleanupSucceeded"] = True + evidence_report["containsProprietaryParentConfiguration"] = False + report = _sanitize_value(report, redactions, redact_tokens=False) + _atomic_write_report(report_target, report) + return exit_code, report + + +def main(argv=None) -> int: + arguments = _argument_parser().parse_args(argv) + try: + exit_code, report = execute_gate( + binary=arguments.binary, + binary_args=arguments.binary_arg, + plugin_root=arguments.plugin_root, + database=arguments.database, + sources=arguments.sources, + parent_configuration=arguments.parent_configuration, + platform_path=arguments.platform_path, + platform_version=arguments.platform_version, + report_path=arguments.report, + evidence_dir=arguments.evidence_dir, + builder=arguments.builder, + db_user=arguments.db_user, + timeout_seconds=arguments.timeout_seconds, + execute=arguments.execute, + allow_empty_password=arguments.allow_empty_password, + ) + except SourceError as error: + print(f"source error: {error}", file=sys.stderr) + return 2 + if exit_code != 0: + failures = report.get("summary", {}).get("failures", []) + detail = "; ".join(str(item) for item in failures) or report.get( + "sourceError", {} + ).get("message", "verification failed") + print(f"issue #76 verification failed: {detail}", file=sys.stderr) + return exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/dev/test_verify_issue_76_roundtrip.py b/tests/dev/test_verify_issue_76_roundtrip.py new file mode 100644 index 00000000..d14b0aff --- /dev/null +++ b/tests/dev/test_verify_issue_76_roundtrip.py @@ -0,0 +1,1729 @@ +import copy +import hashlib +import importlib.util +import json +import os +import subprocess +import tempfile +import time +import unittest +from pathlib import Path +from unittest import mock + + +ROOT = Path(__file__).resolve().parents[2] +SCRIPT = ROOT / "scripts/dev/verify-issue-76-roundtrip.py" + +CATALOG_METADATA_PATH = "Catalog.ЗависимостиСчетов" +MODULE_METADATA_PATH = ( + "CommonModule.СообщенияВСлужбуТехническойПоддержкиБПКлиентСервер.Module" +) +CATALOG_RELATIVE_PATH = Path("src/Catalogs/ЗависимостиСчетов.xml") +MODULE_RELATIVE_PATH = Path( + "src/CommonModules/" + "СообщенияВСлужбуТехническойПоддержкиБПКлиентСервер/Ext/Module.bsl" +) +PARENT_CONFIGURATION_RELATIVE_PATH = Path( + "src/Ext/ParentConfigurations/УправлениеХолдингом.cf" +) +METADATA_MARKER = "UNICA_ISSUE_76_ROUND_TRIP" +BSL_MARKER = f"// {METADATA_MARKER}" + + +def load_verifier(): + if not SCRIPT.is_file(): + raise AssertionError(f"issue #76 verifier implementation is missing: {SCRIPT}") + spec = importlib.util.spec_from_file_location( + "verify_issue_76_roundtrip", + SCRIPT, + ) + if spec is None or spec.loader is None: + raise AssertionError(f"cannot load issue #76 verifier: {SCRIPT}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def sha256(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +def write_workspace(root: Path) -> Path: + workspace = root / "private-workspace" + source = workspace / "src" + catalog = workspace / CATALOG_RELATIVE_PATH + module = workspace / MODULE_RELATIVE_PATH + catalog.parent.mkdir(parents=True) + module.parent.mkdir(parents=True) + (source / "Ext").mkdir(parents=True) + (source / "Configuration.xml").write_text( + "" + "" + "УправлениеХолдингом" + "", + encoding="utf-8", + ) + catalog.write_text( + "" + "" + "ЗависимостиСчетов" + "", + encoding="utf-8", + ) + common_module_descriptor = module.parents[1].with_suffix(".xml") + common_module_descriptor.write_text( + "" + "" + "" + "СообщенияВСлужбуТехническойПоддержкиБПКлиентСервер" + "", + encoding="utf-8", + ) + module.write_bytes( + b"\xef\xbb\xbf" + + ( + "Функция ЛинияПоддержки()\r\n" + "\tВозврат Неопределено;\r\n" + "КонецФункции" + ).encode("utf-8") + ) + (source / "Ext/ParentConfigurations.bin").write_text( + "synthetic support fixture", + encoding="utf-8", + ) + (source / "ConfigDumpInfo.xml").write_text( + "", + encoding="utf-8", + ) + (workspace / "v8project.yaml").write_text( + "format: DESIGNER\n" + "builder: DESIGNER\n" + "source-set:\n" + " - name: main\n" + " type: CONFIGURATION\n" + " path: src\n", + encoding="utf-8", + ) + return workspace + + +def write_packaged_manifest( + plugin_root: Path, + *, + unica_sha256: str, + include_secrets: bool = False, +) -> None: + third_party = plugin_root / "third-party" + third_party.mkdir(parents=True, exist_ok=True) + payload = { + "schemaVersion": 2, + "targetTriple": "aarch64-apple-darwin", + "tools": [ + { + "name": "unica", + "version": "0.12.0", + "sourceCommit": "workspace", + "sourceTag": "workspace", + "sha256": unica_sha256, + }, + { + "name": "v8-runner", + "version": "0.5.1", + "sourceCommit": "7ce1b062843d86644fe55741dbe0ee79f7ca767d", + "sourceTag": "master", + "sha256": "b" * 64, + }, + ], + } + if include_secrets: + payload["privateToken"] = "manifest-top-secret" + payload["tools"][1]["password"] = "manifest-runner-secret" + (third_party / "manifest.json").write_text( + json.dumps(payload, ensure_ascii=False), + encoding="utf-8", + ) + + +def write_gate_inputs(root: Path) -> dict: + fixture = write_workspace(root / "fixture") + database = root / "database" + database.mkdir() + (database / "1Cv8.1CD").write_bytes(b"synthetic infobase") + binary = root / "unica" + binary.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + binary.chmod(0o700) + plugin = root / "plugin" + write_packaged_manifest(plugin, unica_sha256=sha256(binary.read_bytes())) + platform = root / "platform" + platform.mkdir() + parent_configuration = root / "1cv8.cf" + parent_configuration.write_bytes(b"synthetic vendor configuration") + return { + "binary": binary, + "binary_args": [], + "plugin_root": plugin, + "database": database, + "sources": fixture / "src", + "parent_configuration": parent_configuration, + "platform_path": platform, + "platform_version": "8.3.27.2214", + "report_path": root / "report.json", + "builder": "DESIGNER", + "db_user": "Администратор", + "timeout_seconds": 60, + "execute": True, + "allow_empty_password": True, + } + + +class ScriptedClient: + """Small public-tool fake; it never launches 1C or an MCP process.""" + + def __init__( + self, + workspace: Path, + *, + mutate_on_partial: bool = False, + lose_after_full_dump: str | None = None, + full_dump_noop: bool = False, + builds_noop: bool = False, + cdfi_build_changes: set[int] | None = None, + diagnostic: str = "", + ) -> None: + self.workspace = workspace + self.mutate_on_partial = mutate_on_partial + self.lose_after_full_dump = lose_after_full_dump + self.full_dump_noop = full_dump_noop + self.builds_noop = builds_noop + self.cdfi_build_changes = cdfi_build_changes or set() + self.build_count = 0 + self.database_catalog: bytes | None = None + self.database_module: bytes | None = None + self.metadata_marker: str | None = None + self.bsl_marker: str | None = None + self.editing_enabled = False + self.source_editable_paths: set[str] = set() + self.database_editing_enabled = False + self.database_editable_paths: set[str] = set() + self.support_sync_was_marker_free = False + self.diagnostic = diagnostic + self.calls: list[tuple[str, dict]] = [] + + def call(self, name: str, arguments: dict, **_kwargs) -> dict: + arguments = copy.deepcopy(arguments) + self.calls.append((name, arguments)) + + if name == "unica.cf.info": + return { + "ok": True, + "data": { + "support": { + "state": "supported", + "editingEnabled": self.editing_enabled, + } + }, + "errors": [], + } + if name == "unica.support.edit": + if "Capability" in arguments: + if arguments.get("dryRun") is False: + self.editing_enabled = True + data = { + "action": "capability", + "applied": True, + "editingEnabled": True, + "recordsChanged": 0, + } + else: + if arguments.get("dryRun") is False: + self.source_editable_paths.add(arguments["Path"]) + data = { + "action": "objectRule", + "applied": True, + "rule": "editable", + "recordsChanged": 1, + } + return {"ok": True, "data": data, "errors": []} + if name == "unica.meta.edit": + self.metadata_marker = arguments["operations"][0]["values"]["Comment"] + if arguments.get("dryRun") is False: + catalog = self.workspace / CATALOG_RELATIVE_PATH + catalog.write_text( + catalog.read_text(encoding="utf-8").replace( + "", + f"{self.metadata_marker}", + ), + encoding="utf-8", + ) + return { + "ok": True, + "data": { + "changed": True, + "validation": {"status": "passed"}, + }, + "errors": [], + } + if name == "unica.code.patch": + self.bsl_marker = arguments["content"] + module = self.workspace / MODULE_RELATIVE_PATH + before = module.read_bytes() + after = before.replace( + "Функция ЛинияПоддержки()".encode("utf-8"), + (self.bsl_marker + "\r\nФункция ЛинияПоддержки()").encode("utf-8"), + ) + if arguments.get("dryRun") is False: + module.write_bytes(after) + return { + "ok": True, + "data": { + "changed": True, + "preHash": sha256(before), + "postHash": sha256(after), + "affectedTarget": { + "sourceSet": "main", + "metadataPath": MODULE_METADATA_PATH, + }, + "validation": {"status": "passed"}, + }, + "errors": [], + } + if name != "unica.runtime.execute": + raise AssertionError(f"unexpected public tool call: {name}") + + operation = arguments.get("operation") + mode = arguments.get("mode") + if operation == "dump" and mode == "partial": + if self.mutate_on_partial: + module = self.workspace / MODULE_RELATIVE_PATH + module.write_bytes(module.read_bytes() + b"\r\n// partial wrote here") + return { + "ok": False, + "summary": "unica.runtime.execute blocked by source sync guard", + "errors": [ + "applied partial dump requires a divergence-safe merge; " + "wait for alkoleft/v8-runner-rust#30" + ], + } + if operation == "build": + self.build_count += 1 + if arguments.get("fullRebuild") is True: + self.database_editing_enabled = self.editing_enabled + self.database_editable_paths = set(self.source_editable_paths) + self.support_sync_was_marker_free = ( + self.metadata_marker is None and self.bsl_marker is None + ) + elif self.metadata_marker is not None or self.bsl_marker is not None: + required_paths = { + CATALOG_RELATIVE_PATH.as_posix(), + MODULE_RELATIVE_PATH.parents[1].with_suffix(".xml").as_posix(), + } + if not self.database_editing_enabled or not required_paths.issubset( + self.database_editable_paths + ): + return { + "ok": False, + "summary": "configuration build rejected locked database support", + "errors": [ + "editing the target metadata object is forbidden" + ], + } + if not self.builds_noop: + self.database_catalog = ( + self.workspace / CATALOG_RELATIVE_PATH + ).read_bytes() + self.database_module = ( + self.workspace / MODULE_RELATIVE_PATH + ).read_bytes() + if self.build_count in self.cdfi_build_changes: + cdfi = self.workspace / "src/ConfigDumpInfo.xml" + cdfi.write_text( + f"", + encoding="utf-8", + ) + return { + "ok": True, + "summary": "configuration built", + "stdout": self.diagnostic, + "errors": [], + } + if operation == "dump" and mode == "full": + if not self.full_dump_noop: + if self.database_catalog is None or self.database_module is None: + raise AssertionError("full dump requires a preceding build") + (self.workspace / CATALOG_RELATIVE_PATH).write_bytes( + self.database_catalog + ) + (self.workspace / MODULE_RELATIVE_PATH).write_bytes( + self.database_module + ) + if self.lose_after_full_dump == "metadata": + if self.metadata_marker is None: + raise AssertionError("metadata marker was not observed") + catalog = self.workspace / CATALOG_RELATIVE_PATH + catalog.write_text( + catalog.read_text(encoding="utf-8").replace( + f"{self.metadata_marker}", + "", + ), + encoding="utf-8", + ) + if self.lose_after_full_dump == "module": + if self.bsl_marker is None: + raise AssertionError("BSL marker was not observed") + module = self.workspace / MODULE_RELATIVE_PATH + module.write_bytes( + module.read_bytes().replace( + (self.bsl_marker + "\r\n").encode("utf-8"), + b"", + ) + ) + return { + "ok": True, + "summary": "configuration dumped through verified staging", + "stdout": self.diagnostic, + "errors": [], + } + raise AssertionError(f"unexpected runtime arguments: {arguments}") + + +class ScriptedSession(ScriptedClient): + def start(self, required_tools) -> None: + self.required_tools = frozenset(required_tools) + + def close(self) -> None: + return None + + +class Issue76RoundTripTests(unittest.TestCase): + def test_input_state_digest_includes_root_directory_metadata(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + (root / "input.bin").write_bytes(b"payload") + original_mode = root.stat().st_mode & 0o777 + before = verifier._stat_tree_digest(root) + changed_mode = 0o750 if original_mode != 0o750 else 0o700 + root.chmod(changed_mode) + try: + after = verifier._stat_tree_digest(root) + finally: + root.chmod(original_mode) + + self.assertNotEqual(before, after) + + def test_input_state_digest_detects_same_size_write_with_restored_mtime(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + payload = root / "input.bin" + payload.write_bytes(b"AAAA") + metadata = payload.stat() + before = verifier._stat_tree_digest(root) + time.sleep(0.01) + + payload.write_bytes(b"BBBB") + os.utime( + payload, + ns=(metadata.st_atime_ns, metadata.st_mtime_ns), + ) + after = verifier._stat_tree_digest(root) + + self.assertNotEqual(before, after) + + @unittest.skipUnless(os.name == "posix", "the IBCMD isolation wrapper is POSIX-only") + def test_ibcmd_builds_use_private_data_then_full_dump_restores_trusted_platform(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + inputs["builder"] = "IBCMD" + trusted_ibcmd = inputs["platform_path"] / "ibcmd" + invocation_log = root / "ibcmd-arguments.json" + trusted_ibcmd.write_text( + "#!/usr/bin/env python3\n" + "import json\n" + "import sys\n" + "from pathlib import Path\n" + f"Path({str(invocation_log)!r}).write_text(" + "json.dumps(sys.argv[1:]), encoding='utf-8')\n", + encoding="utf-8", + ) + trusted_ibcmd.chmod(0o700) + evidence = root / "evidence" + evidence.mkdir() + sessions = [] + + class PlatformCapturingSession(ScriptedSession): + def __init__(self, workspace): + super().__init__(workspace) + self.runtime_platform_paths = [] + + def call(self, name, arguments, **kwargs): + if name == "unica.runtime.execute": + local = (self.workspace / "v8project.local.yaml").read_text( + encoding="utf-8" + ) + path_line = next( + line + for line in local.splitlines() + if line.startswith(" path: ") + ) + self.runtime_platform_paths.append( + ( + arguments.get("operation"), + arguments.get("mode"), + json.loads(path_line.removeprefix(" path: ")), + ) + ) + return super().call(name, arguments, **kwargs) + + def session_factory(_command, _environment, _timeout, *, cwd): + session = PlatformCapturingSession(cwd) + sessions.append(session) + return session + + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=evidence, + session_factory=session_factory, + ) + + wrapper_platform = ( + evidence.resolve() + / "ibcmd-platform" + / inputs["platform_version"] + ) + private_data = evidence.resolve() / "ibcmd-data" + wrapper = wrapper_platform / "ibcmd" + self.assertTrue(wrapper.is_file(), report) + subprocess.run( + [ + str(wrapper), + "infobase", + "--db-path", + "/private/db", + "config", + "import", + "/private/src", + ], + check=True, + ) + forwarded = json.loads(invocation_log.read_text(encoding="utf-8")) + rejected = subprocess.run( + [ + str(wrapper), + "infobase", + "--data=/shared/profile", + "config", + "import", + "/private/src", + ], + check=False, + capture_output=True, + text=True, + ) + forwarded_after_rejection = json.loads( + invocation_log.read_text(encoding="utf-8") + ) + + self.assertEqual(exit_code, 0, report) + self.assertEqual(len(sessions), 1) + runtime_paths = sessions[0].runtime_platform_paths + build_paths = [ + path + for operation, _mode, path in runtime_paths + if operation == "build" + ] + full_dump_path = next( + path + for operation, mode, path in runtime_paths + if operation == "dump" and mode == "full" + ) + self.assertEqual(build_paths, [str(wrapper_platform), str(wrapper_platform)]) + self.assertEqual(wrapper_platform.name, inputs["platform_version"]) + self.assertEqual(full_dump_path, str(inputs["platform_path"].resolve())) + self.assertEqual(forwarded[:3], ["infobase", "--data", str(private_data)]) + self.assertEqual( + forwarded[3:], + ["--db-path", "/private/db", "config", "import", "/private/src"], + ) + self.assertNotEqual(rejected.returncode, 0) + self.assertIn("--data", rejected.stderr) + self.assertEqual(forwarded_after_rejection, forwarded) + self.assertIs(report["runtimeIsolation"]["privateIbcmdData"], True) + self.assertEqual(report["runtimeIsolation"]["buildDataPath"], "$EVIDENCE/ibcmd-data") + self.assertEqual(report["runtimeIsolation"]["fullDumpPlatformPath"], "$PLATFORM") + + def test_cleanup_failure_is_written_as_terminal_source_error(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + automatic_evidence = root / "automatic-evidence" + automatic_evidence.mkdir() + temporary = mock.Mock() + temporary.name = str(automatic_evidence) + temporary.cleanup.side_effect = OSError("synthetic cleanup failure") + + with mock.patch.object( + verifier.tempfile, + "TemporaryDirectory", + return_value=temporary, + ): + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + session_factory=lambda *_args, cwd, **_kwargs: ScriptedSession( + cwd + ), + ) + + persisted = json.loads(inputs["report_path"].read_text(encoding="utf-8")) + + self.assertEqual(exit_code, 2, report) + self.assertEqual(persisted["status"], "source-error") + self.assertEqual(persisted["exitCode"], 2) + self.assertIs(persisted["summary"]["passed"], False) + self.assertIs(persisted["evidence"]["retained"], True) + self.assertIs(persisted["evidence"]["cleanupSucceeded"], False) + self.assertIn("cleanup", persisted["sourceError"]["message"]) + + def test_session_close_oserror_still_cleans_up_and_writes_terminal_report(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + automatic_evidence = root / "automatic-evidence" + automatic_evidence.mkdir() + temporary = mock.Mock() + temporary.name = str(automatic_evidence) + + class CloseFailingSession(ScriptedSession): + def close(self) -> None: + raise OSError("synthetic MCP close failure") + + with mock.patch.object( + verifier.tempfile, + "TemporaryDirectory", + return_value=temporary, + ): + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + session_factory=lambda *_args, cwd, **_kwargs: CloseFailingSession( + cwd + ), + ) + + persisted = json.loads(inputs["report_path"].read_text(encoding="utf-8")) + + self.assertEqual(exit_code, 2, report) + self.assertEqual(persisted["status"], "source-error") + self.assertEqual(persisted["exitCode"], 2) + self.assertIs(persisted["summary"]["passed"], False) + self.assertIn("close", persisted["sourceError"]["message"]) + temporary.cleanup.assert_called_once_with() + self.assertIs(persisted["evidence"]["retained"], False) + self.assertIs(persisted["evidence"]["cleanupSucceeded"], True) + + def test_unexpected_session_error_still_cleans_up_and_writes_terminal_report(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + automatic_evidence = root / "automatic-evidence" + automatic_evidence.mkdir() + temporary = mock.Mock() + temporary.name = str(automatic_evidence) + + class StartFailingSession(ScriptedSession): + def start(self, _required_tools) -> None: + raise OSError("synthetic unexpected session failure") + + with mock.patch.object( + verifier.tempfile, + "TemporaryDirectory", + return_value=temporary, + ): + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + session_factory=lambda *_args, cwd, **_kwargs: StartFailingSession( + cwd + ), + ) + + persisted = json.loads(inputs["report_path"].read_text(encoding="utf-8")) + + self.assertEqual(exit_code, 2, report) + self.assertEqual(persisted["status"], "source-error") + self.assertIn("unexpected", persisted["sourceError"]["message"]) + temporary.cleanup.assert_called_once_with() + self.assertIs(persisted["evidence"]["retained"], False) + self.assertIs(persisted["evidence"]["cleanupSucceeded"], True) + + def test_unexpected_integrity_probe_error_still_cleans_up_and_writes_report(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + automatic_evidence = root / "automatic-evidence" + automatic_evidence.mkdir() + temporary = mock.Mock() + temporary.name = str(automatic_evidence) + original_digest = verifier._stat_tree_digest + session_started = False + + def flaky_digest(path): + if session_started: + raise RuntimeError("synthetic integrity probe failure") + return original_digest(path) + + class StartFailingSession(ScriptedSession): + def start(self, _required_tools) -> None: + nonlocal session_started + session_started = True + raise verifier.SourceError("synthetic MCP start failure") + + with mock.patch.object( + verifier.tempfile, + "TemporaryDirectory", + return_value=temporary, + ), mock.patch.object( + verifier, + "_stat_tree_digest", + side_effect=flaky_digest, + ): + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + session_factory=lambda *_args, cwd, **_kwargs: StartFailingSession( + cwd + ), + ) + + persisted = json.loads(inputs["report_path"].read_text(encoding="utf-8")) + + self.assertEqual(exit_code, 2, report) + self.assertEqual(persisted["status"], "source-error") + self.assertIn("integrity", persisted["sourceError"]["message"]) + temporary.cleanup.assert_called_once_with() + + def test_unexpected_cleanup_error_is_normalized_into_terminal_report(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + automatic_evidence = root / "automatic-evidence" + automatic_evidence.mkdir() + temporary = mock.Mock() + temporary.name = str(automatic_evidence) + temporary.cleanup.side_effect = RuntimeError( + "synthetic unexpected cleanup failure" + ) + + with mock.patch.object( + verifier.tempfile, + "TemporaryDirectory", + return_value=temporary, + ): + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + session_factory=lambda *_args, cwd, **_kwargs: ScriptedSession( + cwd + ), + ) + + persisted = json.loads(inputs["report_path"].read_text(encoding="utf-8")) + + self.assertEqual(exit_code, 2, report) + self.assertEqual(persisted["status"], "source-error") + self.assertIn("cleanup", persisted["sourceError"]["message"]) + self.assertIs(persisted["evidence"]["retained"], True) + + def test_short_db_user_redaction_does_not_corrupt_report_schema(self): + verifier = load_verifier() + for db_user in ("a", "status", "EVIDENCE", "pass", "main", "/"): + with self.subTest(db_user=db_user), tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + inputs["db_user"] = db_user + + def session_factory(_command, _environment, _timeout, *, cwd): + return ScriptedSession( + cwd, + diagnostic=f"authenticated user {db_user}", + ) + + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + session_factory=session_factory, + ) + persisted = json.loads( + inputs["report_path"].read_text(encoding="utf-8") + ) + + self.assertEqual(exit_code, 0, report) + self.assertEqual(report["status"], "pass") + self.assertEqual(persisted["status"], "pass") + self.assertIs(persisted["summary"]["passed"], True) + rendered = json.dumps(persisted, ensure_ascii=False) + self.assertNotIn("st$DB_USERtus", rendered) + self.assertNotIn("$$DB_USER", rendered) + self.assertIn("authenticated user $DB_USER", rendered) + source_sets = { + step["arguments"].get("sourceSet") + for step in persisted["steps"] + if "sourceSet" in step["arguments"] + } + self.assertEqual(source_sets, {"main"}) + self.assertEqual( + persisted["evidence"]["workspace"], + "$EVIDENCE/workspace", + ) + + def test_secret_environment_is_neither_inherited_nor_serialized(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + secret_environment = { + "AWS_SECRET_ACCESS_KEY": "aws-secret-value-for-issue-76", + "CI_JOB_JWT": "ci-jwt-value-for-issue-76", + "GH_PAT": "github-pat-value-for-issue-76", + "HASP_TOKEN": "hasp-token-value-for-issue-76", + "KUBECONFIG": "/private/credential/kubeconfig-issue-76", + "LC_SECRET": "locale-secret-value-for-issue-76", + "NETHASP_PASSWORD": "nethasp-password-value-for-issue-76", + "SSH_AUTH_SOCK": "/private/credential/ssh-agent-issue-76.sock", + } + hostile_temporary_environment = { + "TMPDIR": "/private/hostile/issue-76-tmpdir", + "TMP": "/private/hostile/issue-76-tmp", + "TEMP": "/private/hostile/issue-76-temp", + } + captured_environment = {} + + def session_factory(_command, environment, _timeout, *, cwd): + captured_environment.update(environment) + diagnostic = "backend said " + " ".join(secret_environment.values()) + return ScriptedSession(cwd, diagnostic=diagnostic) + + with mock.patch.dict( + os.environ, + {**secret_environment, **hostile_temporary_environment}, + clear=False, + ): + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + session_factory=session_factory, + ) + persisted = json.loads(inputs["report_path"].read_text(encoding="utf-8")) + + self.assertEqual(exit_code, 0, report) + rendered = json.dumps(persisted, ensure_ascii=False) + for name, secret in secret_environment.items(): + self.assertNotIn(name, captured_environment) + self.assertNotIn(secret, rendered) + private_work = str(Path(captured_environment["PWD"]) / "work") + for name, hostile_path in hostile_temporary_environment.items(): + self.assertEqual(captured_environment[name], private_work) + self.assertNotEqual(captured_environment[name], hostile_path) + + @unittest.skipUnless(hasattr(os, "symlink"), "symlink support is required") + def test_private_preimage_restore_rejects_a_symlinked_ancestor(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = root / "workspace/src" + outside = root / "outside" + source.mkdir(parents=True) + outside.mkdir() + outside_target = outside / "Item.xml" + outside_before = b"outside sentinel" + outside_target.write_bytes(outside_before) + (source / "Catalogs").symlink_to(outside, target_is_directory=True) + + with self.assertRaises(verifier.SourceError): + verifier._restore_private_preimage( + source / "Catalogs/Item.xml", + b"private preimage", + expected_current_sha256=sha256(outside_before), + ) + + self.assertEqual(outside_target.read_bytes(), outside_before) + + def test_explicit_evidence_source_error_reports_retained_parent_configuration(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + evidence = root / "evidence" + evidence.mkdir() + + class StartFailingSession(ScriptedSession): + def start(self, _required_tools): + original = inputs["sources"] / "Configuration.xml" + original.write_bytes(original.read_bytes() + b" ") + raise verifier.SourceError("synthetic MCP start failure") + + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=evidence, + session_factory=lambda *_args, cwd, **_kwargs: StartFailingSession( + cwd + ), + ) + private_parent = evidence / "workspace" / PARENT_CONFIGURATION_RELATIVE_PATH + private_parent_exists = private_parent.is_file() + + self.assertEqual(exit_code, 2, report) + self.assertEqual(report["status"], "source-error") + self.assertTrue(private_parent_exists) + self.assertIs(report["evidence"]["retained"], True) + self.assertIs(report["evidence"]["cleanupSucceeded"], None) + self.assertIs( + report["evidence"]["containsProprietaryParentConfiguration"], + True, + ) + self.assertIs(report["inputs"]["privateCopiesOnly"], False) + self.assertIn( + "input", + " ".join(report["summary"]["failures"]).casefold(), + ) + + def test_fixed_marker_cannot_false_pass_when_database_was_preseeded(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + workspace = write_workspace(Path(tmp)) + client = ScriptedClient(workspace, builds_noop=True) + catalog = workspace / CATALOG_RELATIVE_PATH + module = workspace / MODULE_RELATIVE_PATH + client.database_catalog = catalog.read_bytes().replace( + b"", + f"{METADATA_MARKER}".encode("utf-8"), + ) + client.database_module = module.read_bytes().replace( + "Функция ЛинияПоддержки()".encode("utf-8"), + (BSL_MARKER + "\r\nФункция ЛинияПоддержки()").encode("utf-8"), + ) + + exit_code, report = verifier.run_roundtrip_flow( + client, + workspace=workspace, + redactions=[(workspace, "$EVIDENCE")], + ) + + self.assertEqual(exit_code, 1, report) + self.assertEqual(report["status"], "failed") + + def test_support_setup_must_transition_from_locked_to_applied_editable_rules(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + workspace = write_workspace(Path(tmp)) + + class NoopSupportClient(ScriptedClient): + def call(self, name, arguments, **kwargs): + if name == "unica.cf.info": + self.calls.append((name, copy.deepcopy(arguments))) + return { + "ok": True, + "data": { + "support": { + "state": "supported", + "editingEnabled": True, + } + }, + "errors": [], + } + if name == "unica.support.edit": + self.calls.append((name, copy.deepcopy(arguments))) + return { + "ok": True, + "data": { + "action": "capability", + "applied": False, + "reason": "already editable", + }, + "errors": [], + } + return super().call(name, arguments, **kwargs) + + client = NoopSupportClient(workspace) + + exit_code, report = verifier.run_roundtrip_flow( + client, + workspace=workspace, + redactions=[(workspace, "$EVIDENCE")], + ) + + self.assertEqual(exit_code, 1, report) + self.assertEqual(report["status"], "failed") + self.assertIn("support", report["summary"]["failures"][0]) + + def test_support_setup_rejects_locked_object_rule_receipt(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + workspace = write_workspace(Path(tmp)) + + class LockedRuleClient(ScriptedClient): + def call(self, name, arguments, **kwargs): + if name == "unica.support.edit" and "Set" in arguments: + self.calls.append((name, copy.deepcopy(arguments))) + return { + "ok": True, + "data": { + "action": "objectRule", + "applied": True, + "rule": "locked", + "recordsChanged": 1, + }, + "errors": [], + } + return super().call(name, arguments, **kwargs) + + client = LockedRuleClient(workspace) + exit_code, report = verifier.run_roundtrip_flow( + client, + workspace=workspace, + redactions=[(workspace, "$EVIDENCE")], + ) + + self.assertEqual(exit_code, 1, report) + self.assertEqual(report["status"], "failed") + self.assertIn("support", report["summary"]["failures"][0]) + called = [name for name, _args in client.calls] + for downstream_tool in ( + "unica.meta.edit", + "unica.code.patch", + "unica.runtime.execute", + ): + self.assertNotIn(downstream_tool, called) + + def test_report_path_cannot_write_into_executable_or_runtime_inputs(self): + verifier = load_verifier() + for protected in ("binary", "manifest", "plugin", "platform"): + with self.subTest(protected=protected), tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + if protected == "binary": + report_path = inputs["binary"] + before = report_path.read_bytes() + elif protected == "manifest": + report_path = inputs["plugin_root"] / "third-party/manifest.json" + before = report_path.read_bytes() + elif protected == "plugin": + report_path = inputs["plugin_root"] / "unsafe-report.json" + before = None + else: + report_path = inputs["platform_path"] / "unsafe-report.json" + before = None + inputs["report_path"] = report_path + + with self.assertRaises(verifier.SourceError): + verifier.execute_gate( + **inputs, + evidence_dir=None, + session_factory=lambda *_args, cwd, **_kwargs: ScriptedSession( + cwd + ), + ) + + if before is None: + self.assertFalse(report_path.exists()) + else: + self.assertEqual(report_path.read_bytes(), before) + + def test_evidence_directory_cannot_overlap_plugin_or_platform_runtime(self): + verifier = load_verifier() + for protected in ("plugin_root", "platform_path"): + with self.subTest(protected=protected), tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + evidence = inputs[protected] / "unsafe-evidence" + evidence.mkdir() + + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=evidence, + session_factory=lambda *_args, cwd, **_kwargs: ScriptedSession( + cwd + ), + ) + + self.assertEqual(exit_code, 2, report) + self.assertEqual(report["status"], "source-error") + self.assertFalse((evidence / "workspace").exists()) + + def test_manifest_unica_digest_must_match_executed_binary(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + write_packaged_manifest( + inputs["plugin_root"], + unica_sha256="a" * 64, + ) + + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + session_factory=lambda *_args, cwd, **_kwargs: ScriptedSession(cwd), + ) + + self.assertEqual(exit_code, 2, report) + self.assertEqual(report["status"], "source-error") + self.assertIn("sha256", report["sourceError"]["message"].casefold()) + + def test_parent_configuration_is_injected_only_into_private_source_copy(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source_copy = write_workspace(root) / "src" + parent_configuration = root / "1cv8.cf" + parent_bytes = b"exact vendor configuration payload" + parent_configuration.write_bytes(parent_bytes) + + receipt = verifier._install_parent_configuration( + parent_configuration, + source_copy, + ) + + destination = ( + source_copy + / PARENT_CONFIGURATION_RELATIVE_PATH.relative_to("src") + ) + self.assertEqual(destination.read_bytes(), parent_bytes) + self.assertEqual(parent_configuration.read_bytes(), parent_bytes) + self.assertEqual(receipt["sha256"], sha256(parent_bytes)) + self.assertEqual(receipt["bytes"], len(parent_bytes)) + + destination.write_bytes(b"existing private payload") + with self.assertRaises(verifier.SourceError): + verifier._install_parent_configuration( + parent_configuration, + source_copy, + ) + self.assertEqual(destination.read_bytes(), b"existing private payload") + + def test_execute_gate_reports_parent_copy_and_proves_input_unchanged(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + evidence = root / "evidence" + evidence.mkdir() + parent_before = inputs["parent_configuration"].read_bytes() + sessions = [] + commands = [] + + def session_factory(command, _environment, _timeout, *, cwd): + commands.append(list(command)) + session = ScriptedSession(cwd) + sessions.append(session) + return session + + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=evidence, + session_factory=session_factory, + ) + + private_parent = evidence / "workspace" / PARENT_CONFIGURATION_RELATIVE_PATH + self.assertEqual(private_parent.read_bytes(), parent_before) + self.assertEqual(inputs["parent_configuration"].read_bytes(), parent_before) + private_binary = evidence.resolve() / "unica-executable" + self.assertEqual(Path(commands[0][0]), private_binary) + self.assertEqual(private_binary.read_bytes(), inputs["binary"].read_bytes()) + self.assertNotEqual(private_binary.stat().st_ino, inputs["binary"].stat().st_ino) + + self.assertEqual(exit_code, 0, report) + self.assertEqual(len(sessions), 1) + self.assertIs(report["provenance"]["executedPrivateBinaryCopy"], True) + parent_report = report["inputs"]["parentConfiguration"] + self.assertEqual(parent_report["path"], "$PARENT_CONFIGURATION_INPUT") + self.assertIs(parent_report["statUnchanged"], True) + self.assertIs(parent_report["hashUnchanged"], True) + self.assertEqual(parent_report["copy"]["sha256"], sha256(parent_before)) + self.assertIs(report["inputs"]["privateCopiesOnly"], True) + self.assertIs( + report["evidence"]["containsProprietaryParentConfiguration"], + True, + ) + + def test_automatic_evidence_cleanup_reports_parent_configuration_absent(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + inputs = write_gate_inputs(Path(tmp)) + evidence_roots = [] + + def session_factory(_command, _environment, _timeout, *, cwd): + evidence_roots.append(Path(cwd).parent) + return ScriptedSession(cwd) + + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + session_factory=session_factory, + ) + persisted = json.loads( + inputs["report_path"].read_text(encoding="utf-8") + ) + + self.assertEqual(exit_code, 0, report) + self.assertEqual(len(evidence_roots), 1) + self.assertFalse(evidence_roots[0].exists()) + self.assertIs(persisted["evidence"]["retained"], False) + self.assertIs(persisted["evidence"]["cleanupSucceeded"], True) + self.assertIs( + persisted["evidence"]["containsProprietaryParentConfiguration"], + False, + ) + + def test_session_factory_failure_does_not_claim_private_binary_execution(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + inputs = write_gate_inputs(root) + evidence = root / "evidence" + evidence.mkdir() + + def session_factory(command, _environment, _timeout, *, cwd): + self.assertEqual( + Path(command[0]), + evidence.resolve() / "unica-executable", + ) + self.assertEqual(cwd, evidence.resolve() / "workspace") + raise verifier.SourceError("synthetic process launch failure") + + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=evidence, + session_factory=session_factory, + ) + + self.assertEqual(exit_code, 2, report) + self.assertEqual(report["status"], "source-error") + self.assertIs( + report["provenance"]["executedPrivateBinaryCopy"], + False, + ) + + def test_cli_requires_and_forwards_explicit_parent_configuration(self): + verifier = load_verifier() + complete = [ + "--binary", + "/private/tmp/unica", + "--plugin-root", + "/private/tmp/plugin", + "--database", + "/private/tmp/input-ib", + "--sources", + "/private/tmp/input-src", + "--parent-configuration", + "/private/tmp/1cv8.cf", + "--platform-path", + "/opt/1cv8/8.3.27.2214", + "--platform-version", + "8.3.27.2214", + "--report", + "/private/tmp/issue-76-report.json", + "--execute", + "--allow-empty-password", + ] + + with mock.patch.object( + verifier, + "execute_gate", + return_value=(0, {"status": "pass"}), + ) as execute: + self.assertEqual(verifier.main(complete), 0) + + self.assertEqual( + execute.call_args.kwargs["parent_configuration"], + Path("/private/tmp/1cv8.cf"), + ) + + def test_automatic_evidence_directory_is_validated_before_any_copy(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + inputs = write_gate_inputs(Path(tmp)) + nested_evidence = inputs["sources"] / "unsafe-tmpdir" + nested_evidence.mkdir() + temporary = mock.Mock() + temporary.name = str(nested_evidence) + + with mock.patch.object( + verifier.tempfile, + "TemporaryDirectory", + return_value=temporary, + ), mock.patch.object( + verifier, + "_copy_regular_tree", + side_effect=AssertionError("copy must not start for unsafe evidence"), + ) as copy_tree: + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + ) + + self.assertEqual(exit_code, 2, report) + self.assertEqual(report["status"], "source-error") + self.assertIn("evidence directory", report["sourceError"]["message"]) + copy_tree.assert_not_called() + + def test_cleanup_failure_before_evidence_validation_does_not_claim_parent_copy(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + inputs = write_gate_inputs(Path(tmp)) + nested_evidence = inputs["sources"] / "unsafe-tmpdir" + nested_evidence.mkdir() + temporary = mock.Mock() + temporary.name = str(nested_evidence) + temporary.cleanup.side_effect = OSError("synthetic cleanup failure") + + with mock.patch.object( + verifier.tempfile, + "TemporaryDirectory", + return_value=temporary, + ): + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + ) + + self.assertEqual(exit_code, 2, report) + self.assertEqual(report["status"], "source-error") + self.assertIs(report["evidence"]["retained"], True) + self.assertIs(report["evidence"]["cleanupSucceeded"], False) + self.assertIs( + report["evidence"]["containsProprietaryParentConfiguration"], + False, + ) + self.assertNotIn("workspace", report["evidence"]) + + def test_unsafe_tmpdir_is_rejected_before_temporary_directory_creation(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + inputs = write_gate_inputs(Path(tmp)) + source_state_before = verifier._stat_tree_digest(inputs["sources"]) + + with mock.patch.object( + verifier.tempfile, + "tempdir", + str(inputs["sources"]), + ): + exit_code, report = verifier.execute_gate( + **inputs, + evidence_dir=None, + session_factory=lambda *_args, cwd, **_kwargs: ScriptedSession( + cwd + ), + ) + + source_state_after = verifier._stat_tree_digest(inputs["sources"]) + + self.assertEqual(exit_code, 0, report) + self.assertEqual(report["status"], "pass") + self.assertEqual(source_state_after, source_state_before) + + def test_preexisting_scenario_marker_is_rejected_before_baseline_build(self): + verifier = load_verifier() + for target in ("metadata", "module"): + with self.subTest(target=target), tempfile.TemporaryDirectory() as tmp: + workspace = write_workspace(Path(tmp)) + if target == "metadata": + path = workspace / CATALOG_RELATIVE_PATH + path.write_text( + path.read_text(encoding="utf-8").replace( + "", + f"{METADATA_MARKER}", + ), + encoding="utf-8", + ) + else: + path = workspace / MODULE_RELATIVE_PATH + path.write_bytes( + path.read_bytes().replace( + "Функция ЛинияПоддержки()".encode("utf-8"), + (BSL_MARKER + "\r\nФункция ЛинияПоддержки()").encode( + "utf-8" + ), + ) + ) + client = ScriptedClient(workspace) + + exit_code, report = verifier.run_roundtrip_flow( + client, + workspace=workspace, + redactions=[(workspace, "$EVIDENCE")], + ) + + self.assertEqual(exit_code, 1, report) + self.assertIn("already present", report["summary"]["failures"][0]) + self.assertEqual(client.calls, []) + + def test_redaction_covers_secret_keys_structured_values_and_plain_db_user(self): + verifier = load_verifier() + diagnostic = ( + "token=ghp_runtime_secret; password: yaml-secret; " + "api_secret=api-secret; " + '\"password\": \"json-secret\"; --api-token cli-token-secret; ' + "authenticated Администратор" + ) + + sanitized = verifier._sanitize_text( + diagnostic, + [("Администратор", "$DB_USER")], + ) + structured = verifier._sanitize_value( + { + "token": "structured-token-secret", + "nested": {"databasePassword": "structured-password-secret"}, + "ordinary": "visible", + }, + [], + ) + step = verifier._step_record( + step_id="redaction", + tool="unica.runtime.execute", + arguments={"token": "step-token-secret"}, + payload={"ok": False, "errors": [diagnostic]}, + duration_ms=1, + redactions=[("Администратор", "$DB_USER")], + ) + + rendered = json.dumps( + {"text": sanitized, "structured": structured, "step": step}, + ensure_ascii=False, + ) + for secret in ( + "ghp_runtime_secret", + "yaml-secret", + "api-secret", + "json-secret", + "cli-token-secret", + "Администратор", + "structured-token-secret", + "structured-password-secret", + "step-token-secret", + ): + self.assertNotIn(secret, rendered) + self.assertEqual(structured["ordinary"], "visible") + self.assertEqual( + step["argumentsSha256"], + verifier._json_digest({"token": ""}), + ) + + def test_packaged_manifest_provenance_is_allowlisted_and_identifies_runner(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + plugin = Path(tmp) / "plugin" + binary_sha256 = "a" * 64 + write_packaged_manifest( + plugin, + unica_sha256=binary_sha256, + include_secrets=True, + ) + + provenance = verifier._packaged_manifest_provenance( + plugin, + unica_binary_sha256=binary_sha256, + ) + + self.assertEqual(provenance["schemaVersion"], 2) + self.assertEqual(provenance["targetTriple"], "aarch64-apple-darwin") + self.assertEqual( + provenance["tools"]["v8-runner"]["sourceCommit"], + "7ce1b062843d86644fe55741dbe0ee79f7ca767d", + ) + self.assertEqual( + set(provenance["tools"]), + {"unica", "v8-runner"}, + ) + self.assertEqual( + set(provenance["tools"]["v8-runner"]), + {"version", "sourceCommit", "sourceTag", "sha256"}, + ) + rendered = json.dumps(provenance, ensure_ascii=False) + self.assertNotIn("manifest-top-secret", rendered) + self.assertNotIn("manifest-runner-secret", rendered) + + def test_input_change_cannot_leave_passed_summary_in_source_error_report(self): + verifier = load_verifier() + report = { + "status": "pass", + "exitCode": 0, + "summary": {"passed": True, "failures": []}, + } + + verifier._record_source_error( + report, + "an input tree changed while the private live scenario ran", + ) + + self.assertEqual((report["status"], report["exitCode"]), ("source-error", 2)) + self.assertIs(report["summary"]["passed"], False) + self.assertEqual( + report["summary"]["failures"], + ["an input tree changed while the private live scenario ran"], + ) + self.assertEqual( + report["sourceError"]["message"], + "an input tree changed while the private live scenario ran", + ) + + def test_ibcmd_project_has_explicit_user_and_empty_password_fields(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + workspace = Path(tmp) + verifier._write_project_configuration( + workspace, + database_copy=workspace / "ib", + platform_path=workspace / "platform", + platform_version="8.3.27.2214", + db_user="Администратор", + builder="IBCMD", + timeout_seconds=7200, + ) + + local = (workspace / "v8project.local.yaml").read_text(encoding="utf-8") + + self.assertIn(' user: "Администратор"\n', local) + self.assertIn(' password: ""\n', local) + self.assertNotIn("Usr=", local) + self.assertNotIn("Pwd=", local) + + def test_cli_requires_both_mutation_opt_ins_before_execute_gate(self): + verifier = load_verifier() + complete = [ + "--binary", + "/private/tmp/unica", + "--plugin-root", + "/private/tmp/plugin", + "--database", + "/private/tmp/input-ib", + "--sources", + "/private/tmp/input-src", + "--parent-configuration", + "/private/tmp/1cv8.cf", + "--platform-path", + "/opt/1cv8/8.3.27.2214", + "--platform-version", + "8.3.27.2214", + "--report", + "/private/tmp/issue-76-report.json", + "--execute", + "--allow-empty-password", + ] + + for omitted in ("--execute", "--allow-empty-password"): + with self.subTest(omitted=omitted), mock.patch.object( + verifier, + "execute_gate", + side_effect=AssertionError("execute_gate must not run"), + ) as execute, mock.patch("sys.stderr"): + argv = [argument for argument in complete if argument != omitted] + with self.assertRaises(SystemExit) as error: + verifier.main(argv) + self.assertEqual(error.exception.code, 2) + execute.assert_not_called() + + def test_scripted_flow_syncs_support_then_uses_ordinary_mutation_build(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + workspace = write_workspace(Path(tmp)) + client = ScriptedClient(workspace) + + exit_code, report = verifier.run_roundtrip_flow( + client, + workspace=workspace, + redactions=[(workspace, "$EVIDENCE")], + ) + + self.assertEqual(exit_code, 0, report) + self.assertEqual((report["status"], report["exitCode"]), ("pass", 0)) + runtime_calls = [ + arguments + for name, arguments in client.calls + if name == "unica.runtime.execute" + ] + builds = [item for item in runtime_calls if item.get("operation") == "build"] + self.assertEqual(len(builds), 2, runtime_calls) + baseline_build, mutation_build = builds + for build in (baseline_build, mutation_build): + self.assertEqual(build["sourceSet"], "main") + self.assertIs(build["dryRun"], False) + self.assertIs(baseline_build["fullRebuild"], True) + self.assertNotIn("fullRebuild", mutation_build) + call_names = [ + (name, arguments.get("operation"), arguments.get("dryRun")) + for name, arguments in client.calls + ] + baseline_index = call_names.index( + ("unica.runtime.execute", "build", False) + ) + support_after_index = call_names.index(("unica.cf.info", None, None), 1) + meta_apply_index = call_names.index(("unica.meta.edit", None, False)) + second_build_index = len(call_names) - 1 - call_names[::-1].index( + ("unica.runtime.execute", "build", False) + ) + self.assertLess(support_after_index, baseline_index) + self.assertLess(baseline_index, meta_apply_index) + self.assertGreater(second_build_index, meta_apply_index) + self.assertEqual(report["builds"]["baselineBuild"]["ok"], True) + self.assertEqual(report["builds"]["mutationBuild"]["ok"], True) + self.assertIs(client.support_sync_was_marker_free, True) + self.assertIs(client.database_editing_enabled, True) + self.assertEqual( + client.database_editable_paths, + { + CATALOG_RELATIVE_PATH.as_posix(), + ( + MODULE_RELATIVE_PATH.parents[1].with_suffix(".xml") + ).as_posix(), + }, + ) + full_dump = next( + item + for item in runtime_calls + if item.get("operation") == "dump" and item.get("mode") == "full" + ) + self.assertEqual(full_dump["sourceSet"], "main") + self.assertIs(full_dump["dryRun"], False) + self.assertEqual(report["roundTrip"]["metadata"]["survived"], True) + self.assertEqual(report["roundTrip"]["module"]["survived"], True) + + def test_noop_full_dump_cannot_pass_source_to_database_to_source_roundtrip(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + workspace = write_workspace(Path(tmp)) + client = ScriptedClient(workspace, full_dump_noop=True) + + exit_code, report = verifier.run_roundtrip_flow( + client, + workspace=workspace, + redactions=[(workspace, "$EVIDENCE")], + ) + + self.assertEqual(exit_code, 1, report) + self.assertEqual(report["status"], "failed") + self.assertIs(report["roundTrip"]["metadata"]["survived"], False) + self.assertIs(report["roundTrip"]["module"]["survived"], False) + + def test_config_dump_info_attributes_only_post_baseline_build_churn(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + workspace = write_workspace(Path(tmp)) + client = ScriptedClient(workspace, cdfi_build_changes={1}) + + exit_code, report = verifier.run_roundtrip_flow( + client, + workspace=workspace, + redactions=[(workspace, "$EVIDENCE")], + ) + + self.assertEqual(exit_code, 0, report) + self.assertIs(report["configDumpInfo"]["changedByBaselineBuild"], True) + self.assertIs(report["configDumpInfo"]["changedByBuild"], False) + + def test_applied_partial_guard_must_refuse_without_mutating_sources(self): + verifier = load_verifier() + for mutate, expected_exit in ((False, 0), (True, 1)): + with self.subTest(mutate=mutate), tempfile.TemporaryDirectory() as tmp: + workspace = write_workspace(Path(tmp)) + client = ScriptedClient(workspace, mutate_on_partial=mutate) + + exit_code, report = verifier.run_roundtrip_flow( + client, + workspace=workspace, + redactions=[(workspace, "$EVIDENCE")], + ) + + self.assertEqual(exit_code, expected_exit, report) + self.assertEqual(report["partialGuard"]["blocked"], True) + self.assertEqual( + report["partialGuard"]["sourceUnchanged"], + not mutate, + ) + if mutate: + operations = [ + arguments.get("operation") + for name, arguments in client.calls + if name == "unica.runtime.execute" + ] + self.assertEqual(operations.count("build"), 1) + self.assertNotIn( + "full", + [ + arguments.get("mode") + for name, arguments in client.calls + if name == "unica.runtime.execute" + ], + ) + + def test_loss_of_either_marker_after_full_dump_is_a_failed_roundtrip(self): + verifier = load_verifier() + for lost in ("metadata", "module"): + with self.subTest(lost=lost), tempfile.TemporaryDirectory() as tmp: + workspace = write_workspace(Path(tmp)) + client = ScriptedClient(workspace, lose_after_full_dump=lost) + + exit_code, report = verifier.run_roundtrip_flow( + client, + workspace=workspace, + redactions=[(workspace, "$EVIDENCE")], + ) + + self.assertEqual(exit_code, 1, report) + self.assertEqual((report["status"], report["exitCode"]), ("failed", 1)) + self.assertEqual(report["roundTrip"][lost]["survived"], False) + + def test_report_redacts_credentials_and_every_absolute_input_path(self): + verifier = load_verifier() + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + workspace = write_workspace(root) + database_input = root / "original database" + source_input = root / "original sources" + database_input.mkdir() + source_input.mkdir() + diagnostic = ( + f'File="{database_input}";Usr="Администратор";' + f'Pwd=super-secret; source={source_input}; work={workspace}' + ) + client = ScriptedClient(workspace, diagnostic=diagnostic) + + exit_code, report = verifier.run_roundtrip_flow( + client, + workspace=workspace, + redactions=[ + (database_input, "$DATABASE_INPUT"), + (source_input, "$SOURCE_INPUT"), + (workspace, "$EVIDENCE"), + ], + ) + + rendered = json.dumps(report, ensure_ascii=False) + self.assertEqual(exit_code, 0, report) + for forbidden in ( + str(database_input), + str(source_input), + str(workspace), + "super-secret", + "Pwd=", + ): + self.assertNotIn(forbidden, rendered) + + +if __name__ == "__main__": + unittest.main()