MCPcopy Create free account
hub / github.com/Keeper-Security/Commander / CommandHandler

Class CommandHandler

keepercommander/service/config/cli_handler.py:20–142  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

18from ...params import KeeperParams
19
20class CommandHandler:
21 @debug_decorator
22 def execute_cli_command(self, params: KeeperParams, command: str) -> str:
23 """Execute a CLI command and capture output."""
24 output = io.StringIO()
25 sys.stdout = output
26 try:
27 # Set service mode flag to bypass master password enforcement
28 original_service_mode = getattr(params, 'service_mode', False)
29 params.service_mode = True
30
31 from ... import cli
32 cli.do_command(params, command)
33 return output.getvalue()
34 except Exception as e:
35 logger.debug(f"Error executing CLI command '{sanitize_command_fields(command)}': {sanitize_debug_data(str(e))}")
36 return ''
37 finally:
38 sys.stdout = sys.__stdout__
39 # Restore original service mode flag
40 params.service_mode = original_service_mode
41
42 @debug_decorator
43 def find_config_record(self, params: KeeperParams, title: str) -> Optional[str]:
44 """Find owned config record by exact title match using vault search.
45
46 Only records owned by the current account are eligible. Shared or
47 non-owned records (even with a matching title) are ignored so that
48 service configuration cannot be supplied or overwritten by another user.
49 """
50 try:
51 from ... import vault_extensions
52
53 logger.debug(f"Searching for record with exact title: '{title}'")
54 records = list(vault_extensions.find_records(params, title))
55
56 for record in records:
57 logger.debug(f"Checking record: '{record.title}' (UID: {record.record_uid})")
58 if record.title != title:
59 continue
60
61 owner = (params.record_owner_cache or {}).get(record.record_uid)
62 if not owner or not owner.owner:
63 logger.debug(
64 f"Skipping non-owned config record '{title}' "
65 f"(UID: {record.record_uid})"
66 )
67 continue
68
69 logger.debug(f"✓ Found owned exact title match: '{title}' (UID: {record.record_uid})")
70 return record.record_uid
71
72 logger.debug(f"✗ No owned record found with exact title: '{title}'")
73 return None
74
75 except Exception as e:
76 logger.error(f"Error searching for record: {e}")
77 print(f"Error searching for record: {e}")

Calls

no outgoing calls