D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
usr
/
lib
/
python3
/
dist-packages
/
uaclient
/
files
/
Filename :
user_config_file.py
back
Copy
import logging import os from typing import Optional from urllib.parse import urlparse from uaclient import defaults, event_logger, util from uaclient.data_types import ( BoolDataValue, DataObject, EnumDataValue, Field, IntDataValue, StringDataValue, ) from uaclient.files.data_types import DataObjectFile, DataObjectFileFormat from uaclient.files.files import UAFile # Config proxy fields that are visible and configurable PROXY_FIELDS = [ "apt_http_proxy", "apt_https_proxy", "global_apt_http_proxy", "global_apt_https_proxy", "ua_apt_http_proxy", "ua_apt_https_proxy", "http_proxy", "https_proxy", ] class LXDGuestAttachEnum(EnumDataValue): ON = "on" OFF = "off" AVAILABLE = "available" def __str__(self): return self.value class UserConfigData(DataObject): fields = [ Field("apt_http_proxy", StringDataValue, required=False), Field("apt_https_proxy", StringDataValue, required=False), Field("global_apt_http_proxy", StringDataValue, required=False), Field("global_apt_https_proxy", StringDataValue, required=False), Field("ua_apt_http_proxy", StringDataValue, required=False), Field("ua_apt_https_proxy", StringDataValue, required=False), Field("http_proxy", StringDataValue, required=False), Field("https_proxy", StringDataValue, required=False), Field("apt_news", BoolDataValue, required=False), Field("apt_news_url", StringDataValue, required=False), Field("poll_for_pro_license", BoolDataValue, required=False), Field("polling_error_retry_delay", IntDataValue, required=False), Field("metering_timer", IntDataValue, required=False), Field("update_messaging_timer", IntDataValue, required=False), Field( "vulnerability_data_url_prefix", StringDataValue, required=False ), Field("lxd_guest_attach", LXDGuestAttachEnum, required=False), ] def __init__( self, apt_http_proxy: Optional[str] = None, apt_https_proxy: Optional[str] = None, global_apt_http_proxy: Optional[str] = None, global_apt_https_proxy: Optional[str] = None, ua_apt_http_proxy: Optional[str] = None, ua_apt_https_proxy: Optional[str] = None, http_proxy: Optional[str] = None, https_proxy: Optional[str] = None, apt_news: Optional[bool] = None, apt_news_url: Optional[str] = None, poll_for_pro_license: Optional[bool] = None, polling_error_retry_delay: Optional[int] = None, metering_timer: Optional[int] = None, update_messaging_timer: Optional[int] = None, vulnerability_data_url_prefix: Optional[str] = None, lxd_guest_attach: Optional[LXDGuestAttachEnum] = None, ): self.apt_http_proxy = apt_http_proxy self.apt_https_proxy = apt_https_proxy self.global_apt_http_proxy = global_apt_http_proxy self.global_apt_https_proxy = global_apt_https_proxy self.ua_apt_http_proxy = ua_apt_http_proxy self.ua_apt_https_proxy = ua_apt_https_proxy self.http_proxy = http_proxy self.https_proxy = https_proxy self.apt_news = apt_news self.apt_news_url = apt_news_url self.poll_for_pro_license = poll_for_pro_license self.polling_error_retry_delay = polling_error_retry_delay self.metering_timer = metering_timer self.update_messaging_timer = update_messaging_timer self.vulnerability_data_url_prefix = vulnerability_data_url_prefix self.lxd_guest_attach = lxd_guest_attach event = event_logger.get_event_logger() LOG = logging.getLogger(util.replace_top_level_logger_name(__name__)) class UserConfigFileObject: def __init__(self, directory: str = defaults.DEFAULT_DATA_DIR): file_name = defaults.USER_CONFIG_FILE self._private = DataObjectFile( UserConfigData, UAFile( file_name, os.path.join(directory, defaults.PRIVATE_SUBDIR), private=True, ), DataObjectFileFormat.JSON, optional_type_errors_become_null=True, ) self._public = DataObjectFile( UserConfigData, UAFile(file_name, directory, private=False), DataObjectFileFormat.JSON, optional_type_errors_become_null=True, ) @property def public_config(self) -> UserConfigData: public_config = self._public.read() if public_config is None: public_config = UserConfigData() return public_config def redact_config_data( self, user_config: UserConfigData ) -> UserConfigData: redacted_data_dict = user_config.to_dict() for field in PROXY_FIELDS: value = redacted_data_dict.get(field) if value: parsed_url = urlparse(value) if parsed_url.username or parsed_url.password: redacted_data_dict[field] = "<REDACTED>" return UserConfigData.from_dict(redacted_data_dict) def read(self) -> UserConfigData: if util.we_are_currently_root(): private_config = self._private.read() if private_config is not None: return private_config public_config = self._public.read() if public_config is not None: return public_config return UserConfigData() def write(self, content: UserConfigData): self._private.write(content) redacted_content = self.redact_config_data(content) self._public.write(redacted_content) user_config = UserConfigFileObject(defaults.DEFAULT_DATA_DIR)