Base class for Docker setup with reusable core logic
| 36 | |
| 37 | |
| 38 | class DockerSetupBase: |
| 39 | """Base class for Docker setup with reusable core logic""" |
| 40 | |
| 41 | @staticmethod |
| 42 | def _is_owned_record(params, record_uid: str) -> bool: |
| 43 | """True when record_owner_cache marks the record as owned by this account.""" |
| 44 | if not record_uid: |
| 45 | return False |
| 46 | owner_cache = getattr(params, 'record_owner_cache', None) or {} |
| 47 | owner = owner_cache.get(record_uid) |
| 48 | return bool(owner and owner.owner) |
| 49 | |
| 50 | @staticmethod |
| 51 | def _is_owned_shared_folder(params, folder_uid: str) -> bool: |
| 52 | """True when the shared folder is owned by the current account.""" |
| 53 | if not folder_uid: |
| 54 | return False |
| 55 | shared_folder = (getattr(params, 'shared_folder_cache', None) or {}).get(folder_uid) |
| 56 | if not shared_folder: |
| 57 | return False |
| 58 | |
| 59 | owner_uid = shared_folder.get('owner_account_uid') |
| 60 | account_uid_bytes = getattr(params, 'account_uid_bytes', None) |
| 61 | if owner_uid and account_uid_bytes: |
| 62 | return owner_uid == utils.base64_url_encode(account_uid_bytes) |
| 63 | |
| 64 | owner_username = shared_folder.get('owner_username') |
| 65 | if owner_username and getattr(params, 'user', None): |
| 66 | return owner_username.lower() == params.user.lower() |
| 67 | return False |
| 68 | |
| 69 | @staticmethod |
| 70 | def _find_owned_ksm_app_by_title(params, app_name: str): |
| 71 | """Return an owned KSM app with this title, or None (ignores non-owned squats).""" |
| 72 | if not app_name: |
| 73 | return None |
| 74 | for rec_cache_val in (getattr(params, 'record_cache', None) or {}).values(): |
| 75 | rec = KSMCommand._app_record_from_cache_entry(rec_cache_val) |
| 76 | if not rec: |
| 77 | continue |
| 78 | try: |
| 79 | title = json.loads(rec['data_unencrypted'].decode('utf-8')).get('title') |
| 80 | except Exception: |
| 81 | continue |
| 82 | if title == app_name and DockerSetupBase._is_owned_record(params, rec.get('record_uid')): |
| 83 | return rec |
| 84 | return None |
| 85 | |
| 86 | @staticmethod |
| 87 | def resolve_commander_config_path(config_path: str = None, params=None) -> str: |
| 88 | """Resolve config.json using the same rules as login/shell startup.""" |
| 89 | if config_path: |
| 90 | resolved = os.path.expanduser(config_path) |
| 91 | elif params and getattr(params, 'config_filename', None): |
| 92 | resolved = os.path.expanduser(params.config_filename) |
| 93 | else: |
| 94 | resolved = str(utils.get_default_path() / 'config.json') |
| 95 | return os.path.abspath(resolved) |
no outgoing calls