Skip to content

Session

thehive4py.session

DEFAULT_RETRY = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504], allowed_methods=['GET', 'POST', 'PUT', 'PATCH', 'DELETE'], raise_on_status=False) module-attribute

RetryValue = Union[Retry, int, None] module-attribute

VerifyValue = Union[bool, str] module-attribute

TheHiveSession(url, apikey=None, username=None, password=None, verify=True, max_retries=DEFAULT_RETRY)

Bases: Session

Source code in thehive4py/session.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(
    self,
    url: str,
    apikey: Optional[str] = None,
    username: Optional[str] = None,
    password: Optional[str] = None,
    verify: VerifyValue = True,
    max_retries: RetryValue = DEFAULT_RETRY,
):
    super().__init__()
    self.hive_url = self._sanitize_hive_url(url)
    self.verify = verify
    self.headers["User-Agent"] = f"thehive4py/{__version__}"
    self._set_retries(max_retries=max_retries)

    if username and password:
        self.headers["Authorization"] = requests.auth._basic_auth_str(
            username, password
        )
    elif apikey:
        self.headers["Authorization"] = f"Bearer {apikey}"
    else:
        raise TheHiveError(
            "Either apikey or the username/password combination must be provided!"
        )

hive_url = self._sanitize_hive_url(url) instance-attribute

verify = verify instance-attribute

make_request(method, path, params=None, data=None, json=None, files=None, download_path=None)

Source code in thehive4py/session.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def make_request(
    self,
    method: str,
    path: str,
    params=None,
    data=None,
    json=None,
    files=None,
    download_path: Union[str, PathLike, None] = None,
) -> Any:
    endpoint_url = f"{self.hive_url}{path}"

    headers = {**self.headers}
    if json is not None:
        data = jsonlib.dumps(json, cls=_SessionJSONEncoder)
        headers = {**headers, "Content-Type": "application/json"}

    response = self.request(
        method,
        url=endpoint_url,
        params=params,
        data=data,
        files=files,
        headers=headers,
        verify=self.verify,
        stream=bool(download_path),
    )

    return self._process_response(response, download_path=download_path)