Gecko [ oric.aiou.edu.pk ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
api	[ DIR ]	drwxr-xr-x
cli	[ DIR ]	drwxr-xr-x
clouds	[ DIR ]	drwxr-xr-x
daemon	[ DIR ]	drwxr-xr-x
entitlements	[ DIR ]	drwxr-xr-x
files	[ DIR ]	drwxr-xr-x
http	[ DIR ]	drwxr-xr-x
messages	[ DIR ]	drwxr-xr-x
timer	[ DIR ]	drwxr-xr-x
__init__.py	0 B	-rw-r--r--
actions.py	14.25 KB	-rw-r--r--
apt.py	34.27 KB	-rw-r--r--
apt_news.py	8.32 KB	-rw-r--r--
config.py	17.36 KB	-rw-r--r--
contract.py	29.96 KB	-rw-r--r--
contract_data_types.py	9.89 KB	-rw-r--r--
data_types.py	10.48 KB	-rw-r--r--
defaults.py	2.52 KB	-rw-r--r--
event_logger.py	8.06 KB	-rw-r--r--
exceptions.py	17.17 KB	-rw-r--r--
gpg.py	836 B	-rw-r--r--
livepatch.py	12.85 KB	-rw-r--r--
lock.py	4.42 KB	-rw-r--r--
log.py	4.69 KB	-rw-r--r--
secret_manager.py	648 B	-rw-r--r--
security_status.py	25.48 KB	-rw-r--r--
snap.py	7.09 KB	-rw-r--r--
status.py	28.42 KB	-rw-r--r--
system.py	26.03 KB	-rw-r--r--
types.py	308 B	-rw-r--r--
upgrade_lts_contract.py	3.54 KB	-rw-r--r--
util.py	15.45 KB	-rw-r--r--
version.py	2.62 KB	-rw-r--r--
yaml.py	840 B	-rw-r--r--

Code Editor : log.py

import json
import logging
import os
import pathlib
from collections import OrderedDict
from typing import Any, Dict, List, Union  # noqa: F401

from uaclient import defaults, secret_manager, system, util
from uaclient.config import UAConfig

class RegexRedactionFilter(logging.Filter):
    """A logging filter to redact confidential info"""

def filter(self, record: logging.LogRecord):
        record.msg = util.redact_sensitive_logs(str(record.msg))
        return True

class KnownSecretRedactionFilter(logging.Filter):
    """A logging filter to redact confidential info"""

def filter(self, record: logging.LogRecord):
        record.msg = secret_manager.secrets.redact_secrets(str(record.msg))
        return True

class JsonArrayFormatter(logging.Formatter):
    """Json Array Formatter for our logging mechanism
    Custom made for Pro logging needs
    """

default_time_format = "%Y-%m-%dT%H:%M:%S"
    default_msec_format = "%s.%03d"
    required_fields = (
        "asctime",
        "levelname",
        "name",
        "funcName",
        "lineno",
        "message",
    )

def format(self, record: logging.LogRecord) -> str:
        record.message = record.getMessage()
        record.asctime = self.formatTime(record)

extra_message_dict = {}  # type: Dict[str, Any]
        if record.exc_info:
            extra_message_dict["exc_info"] = self.formatException(
                record.exc_info
            )
        if not extra_message_dict.get("exc_info") and record.exc_text:
            extra_message_dict["exc_info"] = record.exc_text
        if record.stack_info:
            extra_message_dict["stack_info"] = self.formatStack(
                record.stack_info
            )
        extra = record.__dict__.get("extra")
        if extra and isinstance(extra, dict):
            extra_message_dict.update(extra)

# is ordered to maintain order of fields in log output
        local_log_record = OrderedDict()  # type: Dict[str, Any]
        # update the required fields in the order stated
        for field in self.required_fields:
            value = record.__dict__.get(field)
            local_log_record[field] = value

local_log_record["extra"] = extra_message_dict
        return json.dumps(list(local_log_record.values()))

def get_user_or_root_log_file_path() -> str:
    """
    Gets the correct log_file path,
    adjusting for whether the user is root or not.
    """
    if util.we_are_currently_root():
        return UAConfig().log_file
    else:
        return get_user_log_file()

def get_user_log_file() -> str:
    """Gets the correct user log_file storage location"""
    return os.path.join(system.get_user_cache_dir(), "ubuntu-pro.log")

def get_all_user_log_files() -> List[str]:
    """Gets all the log files for the users in the system

Returns a list of all user log files in their home directories.
    """
    user_directories = os.listdir("/home")
    log_files = []
    for user_directory in user_directories:
        user_path = os.path.join(
            "/home",
            user_directory,
            ".cache",
            defaults.USER_CACHE_SUBDIR,
            "ubuntu-pro.log",
        )
        if os.path.isfile(user_path):
            log_files.append(user_path)
    return log_files

def setup_journald_logging():
    logger = logging.getLogger("ubuntupro")
    logger.setLevel(logging.INFO)
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(JsonArrayFormatter())
    console_handler.setLevel(logging.INFO)
    console_handler.addFilter(RegexRedactionFilter())
    console_handler.addFilter(KnownSecretRedactionFilter())
    logger.addHandler(console_handler)

def setup_cli_logging(log_level: Union[str, int], log_file: str):
    """Setup logging to log_file

If run as non-root then log_file is replaced with a user-specific log file.
    """
    # support lower-case log_level config value
    if isinstance(log_level, str):
        log_level = log_level.upper()

# if we are running as non-root, change log file
    if not util.we_are_currently_root():
        log_file = get_user_log_file()

logger = logging.getLogger("ubuntupro")
    logger.setLevel(log_level)

# Clear all handlers, so they are replaced for this logger
    logger.handlers = []

# Setup file logging
    log_file_path = pathlib.Path(log_file)
    if not log_file_path.exists():
        log_file_path.parent.mkdir(parents=True, exist_ok=True)
        log_file_path.touch(mode=0o640)
    file_handler = logging.FileHandler(log_file)
    file_handler.setFormatter(JsonArrayFormatter())
    file_handler.setLevel(log_level)
    file_handler.addFilter(RegexRedactionFilter())
    file_handler.addFilter(KnownSecretRedactionFilter())

logger.addHandler(file_handler)

${this.title}

Code Editor : log.py