File: session_holder.py

package info (click to toggle)
python-proton-vpn-api-core 4.16.0-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,312 kB
  • sloc: python: 11,057; makefile: 9
file content (146 lines) | stat: -rw-r--r-- 4,959 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
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
62
63
64
65
66
67
68
69
70
71
72
73
74
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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
"""
Proton VPN Session API.


Copyright (c) 2023 Proton AG

This file is part of Proton VPN.

Proton VPN is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

Proton VPN is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with ProtonVPN.  If not, see <https://www.gnu.org/licenses/>.
"""
from __future__ import annotations
from dataclasses import dataclass

from pathlib import Path
import platform
from typing import Optional

import distro

from proton.sso import ProtonSSO
from proton.vpn import logging
from proton.vpn.connection import VPNCredentials
from proton.vpn.session import VPNSession
from proton.vpn.session.utils import to_semver_build_metadata_format, get_core_api_semver_version

logger = logging.getLogger(__name__)

CPU_ARCHITECTURE = to_semver_build_metadata_format(platform.machine())
DISTRIBUTION_ID = distro.id()
DISTRIBUTION_VERSION = distro.version()


def _is_beta_repo_installed() -> bool:
    if distro.id() == "debian" or distro.like() == "debian":
        return Path("/etc/apt/sources.list.d/protonvpn-beta.sources").is_file()

    if distro.id() == "fedora" or distro.like() == "fedora":
        return Path("/etc/yum.repos.d/protonvpn-beta.repo").is_file()

    return False


BETA_REPO_INSTALLED = _is_beta_repo_installed()


@dataclass
class ClientTypeMetadata:  # pylint: disable=missing-class-docstring
    type: str
    version: str = get_core_api_semver_version()


class SessionHolder:
    """Holds the current session object, initializing it lazily when requested."""

    def __init__(
        self, client_type_metadata: ClientTypeMetadata,
        session: VPNSession = None
    ):
        self._proton_sso = ProtonSSO(
            appversion=self._get_app_version_header_value(client_type_metadata),
            user_agent=f"ProtonVPN/{client_type_metadata.version} "
                       f"(Linux; {DISTRIBUTION_ID}/{DISTRIBUTION_VERSION})"
        )
        self._session = session

    def get_session_for(self, username: str) -> VPNSession:
        """
        Returns the session for the specified user.
        :param username: Proton account username.
        :return:
        """
        self._session = self._proton_sso.get_session(
            account_name=username,
            override_class=VPNSession
        )
        return self._session

    @property
    def session(self) -> VPNSession:
        """Returns the current session object."""
        if not self._session:
            self._session = self._proton_sso.get_default_session(
                override_class=VPNSession
            )

        return self._session

    @property
    def user_tier(self) -> Optional[int]:
        """Returns the user tier, if the session is already loaded."""
        if self.session.loaded:
            return self.session.vpn_account.max_tier

        return None

    @property
    def vpn_credentials(self) -> Optional[VPNCredentials]:
        """Returns the VPN credentials, if the session is already loaded."""
        if self.session.loaded:
            return self.session.vpn_account.vpn_credentials

        return None

    @classmethod
    def _get_app_version_header_value(cls, client_type_metadata: ClientTypeMetadata) -> str:
        app_version = f"linux-vpn-{client_type_metadata.type}@{client_type_metadata.version}"

        version_metadata = cls._get_version_metadata()
        if version_metadata:
            app_version += f"+{version_metadata}"

        return app_version

    @staticmethod
    def _get_version_metadata():
        """
        The following version metadata is sent to the REST API for the following reasons:
        1. We want to have an idea on the amount of total users running on each CPU architecture,
           to know which archs we should target when compiling components written in rust.
        2. We want to know if users have our early release linux repositories installed, so that
           2.a We have an idea of the amount of early release users.
           2.b We can enable feature flags only for early release users (i.e. users with our
               official beta release package installed).

        The reason why we do this unconventional use of the semver build metadata section is
        that currently our infra doesn't allow passing/parsing this information through another
        header that's not x-pm-appversion.
        """
        metas = []
        if CPU_ARCHITECTURE:
            metas.append(CPU_ARCHITECTURE)
        if BETA_REPO_INSTALLED:
            metas.append("beta")

        return ".".join(metas)