403WebShell

403Webshell

Server IP : 13.127.148.211 / Your IP : 216.73.216.13
Web Server : Apache/2.4.41 (Ubuntu)
System : Linux ip-172-31-43-195 5.15.0-1084-aws #91~20.04.1-Ubuntu SMP Fri May 2 06:59:36 UTC 2025 x86_64
User : www-data ( 33)
PHP Version : 7.4.3-4ubuntu2.29
Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare,
MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON
Directory : /lib/python3/dist-packages/uaclient/

Upload File :

[ Back ]

Current File : /lib/python3/dist-packages/uaclient/serviceclient.py

import abc
import json
import os
from posixpath import join as urljoin
from typing import Any, Dict, Optional, Tuple, Type
from urllib import error
from urllib.parse import urlencode

from uaclient import config, exceptions, system, util, version


class UAServiceClient(metaclass=abc.ABCMeta):

    url_timeout = 30  # type: Optional[int]
    # Cached serviceclient_url_responses if provided in uaclient.conf
    # via features: {serviceclient_url_responses: /some/file.json}
    _response_overlay = None  # type: Dict[str, Any]

    @property
    @abc.abstractmethod
    def api_error_cls(self) -> Type[Exception]:
        """Set in subclasses to the type of API error raised"""
        pass

    @property
    @abc.abstractmethod
    def cfg_url_base_attr(self) -> str:
        """String in subclasses, the UAConfig attribute containing base url"""
        pass

    def __init__(self, cfg: Optional[config.UAConfig] = None) -> None:
        if not cfg:
            root_mode = os.getuid() == 0
            self.cfg = config.UAConfig(root_mode=root_mode)
        else:
            self.cfg = cfg

    def headers(self):
        return {
            "user-agent": "UA-Client/{}".format(version.get_version()),
            "accept": "application/json",
            "content-type": "application/json",
        }

    def request_url(
        self,
        path,
        data=None,
        headers=None,
        method=None,
        query_params=None,
        potentially_sensitive: bool = True,
        timeout: Optional[int] = None,
    ):
        path = path.lstrip("/")
        if not headers:
            headers = self.headers()
        if headers.get("content-type") == "application/json" and data:
            data = json.dumps(data).encode("utf-8")
        url = urljoin(getattr(self.cfg, self.cfg_url_base_attr), path)
        fake_response, fake_headers = self._get_fake_responses(url)
        if fake_response:
            return fake_response, fake_headers  # URL faked by uaclient.conf
        if query_params:
            # filter out None values
            filtered_params = {
                k: v for k, v in sorted(query_params.items()) if v is not None
            }
            url += "?" + urlencode(filtered_params)
        timeout_to_use = timeout if timeout is not None else self.url_timeout
        try:
            response, headers = util.readurl(
                url=url,
                data=data,
                headers=headers,
                method=method,
                timeout=timeout_to_use,
                potentially_sensitive=potentially_sensitive,
            )
        except error.URLError as e:
            body = None
            if hasattr(e, "body"):
                body = e.body  # type: ignore
            elif hasattr(e, "read"):
                body = e.read().decode("utf-8")  # type: ignore
            if body:
                try:
                    error_details = json.loads(
                        body, cls=util.DatetimeAwareJSONDecoder
                    )
                except ValueError:
                    error_details = None
                if error_details:
                    raise self.api_error_cls(e, error_details)
            raise exceptions.UrlError(
                e, code=getattr(e, "code", None), headers=headers, url=url
            )
        return response, headers

    def _get_response_overlay(self, url: str):
        """Return a list of fake response dicts for a given URL.

        serviceclient_url_responses in uaclient.conf should be a path
        to a json file which contains a dictionary keyed by full URL path.
        Each value will be a list of dicts representing each faked response
        for the given URL.

            The response dict item will have a code: <HTTP_STATUS_CODE> and
               response: "some string of content".
            The JSON string below fakes the available_resources URL on the
            contract server:
            '{"https://contracts.canonical.com/v1/resources": \
               [{"code": 200, "response": {"key": "val1", "key2": "val2"}}]}'

        :return: List of dicts for each faked response matching the url, or
           and empty list when no matching url found.
        """
        if self._response_overlay is not None:
            # Cache it so we don't re-read config every readurl call
            return self._response_overlay.get(url, [])
        response_overlay_path = self.cfg.features.get(
            "serviceclient_url_responses"
        )
        if not response_overlay_path:
            self._response_overlay = {}
        elif not os.path.exists(response_overlay_path):
            self._response_overlay = {}
        else:
            self._response_overlay = json.loads(
                system.load_file(response_overlay_path)
            )
        return self._response_overlay.get(url, [])

    def _get_fake_responses(
        self, url: str
    ) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, str]]]:
        """Return response and headers if faked for this URL in uaclient.conf.

        :return: A tuple of response and header dicts if the URL has an overlay
            response defined. Return (None, {}) otherwise.

        :raises exceptions.URLError: When faked response "code" is != 200.
            URLError reason will be "response" value and any optional
            "headers" provided.
        """
        responses = self._get_response_overlay(url)
        if not responses:
            return None, {}
        if len(responses) == 1:
            # When only one respose is defined, repeat it for all calls
            response = responses[0]
        else:
            # When multiple responses defined pop the first one off the list.
            response = responses.pop(0)
        if response["code"] == 200:
            return response["response"], response.get("headers", {})
        # Must be an error
        e = error.URLError(response["response"])
        url_exception = exceptions.UrlError(
            e,
            code=response["code"],
            headers=response.get("headers", {}),
            url=url,
        )

        if response.get("type") == "contract":
            raise exceptions.ContractAPIError(
                url_exception,
                error_response=response["response"],
            )
        else:
            raise url_exception

Youez - 2016 - github.com/yon3zu
LinuXploit