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
147
|
"""
VPN connection events to react to.
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 typing import TYPE_CHECKING, Optional, Any
from .enum import StateMachineEventEnum
if TYPE_CHECKING:
from proton.vpn.connection.vpnconnection import VPNConnection
@dataclass
class ConnectionDetails:
"""Connection details obtained via local agent."""
device_ip: Optional[str] = None
device_country: Optional[str] = None
server_ipv4: Optional[str] = None
server_ipv6: Optional[str] = None
# pylint: disable=too-few-public-methods
@dataclass
class EventContext:
"""
Relevant event context.
Args:
connection: the VPN connection object that emitted this event.
reason: optional backend-dependent data providing more context about the event.
error: an optional exception to be bubbled up while processing the event.
"""
connection: "VPNConnection"
connection_details: Optional[ConnectionDetails] = None
forwarded_port: Optional[int] = None
reason: Optional[Any] = None
error: Optional[Exception] = None
class Event:
"""Base event that all the other events should inherit from."""
type = None
def __init__(self, context: EventContext = None):
if self.type is None:
raise AttributeError("event attribute not defined")
self.context = context or EventContext(connection=None)
def check_for_errors(self):
"""Raises an exception if there is one."""
if self.context.error:
raise self.context.error
class Initialized(Event):
"""Event that leads to the initial state."""
type = StateMachineEventEnum.INITIALIZED
class Up(Event):
"""Signals that the VPN connection should be started."""
type = StateMachineEventEnum.UP
class Down(Event):
"""Signals that the VPN connection should be stopped."""
type = StateMachineEventEnum.DOWN
class Connected(Event):
"""Signals that the VPN connection was successfully established."""
type = StateMachineEventEnum.CONNECTED
class Disconnected(Event):
"""Signals that the VPN connection was successfully disconnected by the user."""
type = StateMachineEventEnum.DISCONNECTED
class Error(Event):
"""Parent class for events signaling VPN disconnection."""
class DeviceDisconnected(Error):
"""Signals that the VPN connection dropped unintentionally."""
type = StateMachineEventEnum.DEVICE_DISCONNECTED
class Timeout(Error):
"""Signals that a timeout occurred while trying to establish the VPN
connection."""
type = StateMachineEventEnum.TIMEOUT
class AuthDenied(Error):
"""Signals that an authentication denied occurred while trying to establish
the VPN connection."""
type = StateMachineEventEnum.AUTH_DENIED
class ExpiredCertificate(Error):
"""Signals that the passed certificate has expired and needs to be refreshed."""
type = StateMachineEventEnum.CERTIFICATE_EXPIRED
class MaximumSessionsReached(Error):
"""Signals that for the given plan the user has too many devices/sessions connected."""
type = StateMachineEventEnum.MAXIMUM_SESSIONS_REACHED
class TunnelSetupFailed(Error):
"""Signals that there was an error setting up the VPN tunnel."""
type = StateMachineEventEnum.TUNNEL_SETUP_FAILED
class UnexpectedError(Error):
"""Signals that an unexpected error occurred."""
type = StateMachineEventEnum.UNEXPECTED_ERROR
_event_types = [
event_type for event_type in Event.__subclasses__()
if event_type is not Error # As error is an abstract class.
]
_event_types.extend(Error.__subclasses__())
EVENT_TYPES = tuple(_event_types)
|
