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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
"""Login flow v2 API wrapper."""
import asyncio
import json
import time
from dataclasses import dataclass
import httpx
from ._exceptions import check_error
from ._session import AsyncNcSession, NcSession
MAX_TIMEOUT = 60 * 20
@dataclass
class LoginFlow:
"""The Nextcloud Login flow v2 initialization response representation."""
def __init__(self, raw_data: dict) -> None:
self.raw_data = raw_data
@property
def login(self) -> str:
"""The URL for user authorization.
Should be opened by the user in the default browser to authorize in Nextcloud.
"""
return self.raw_data["login"]
@property
def token(self) -> str:
"""Token for a polling for confirmation of user authorization."""
return self.raw_data["poll"]["token"]
@property
def endpoint(self) -> str:
"""Endpoint for polling."""
return self.raw_data["poll"]["endpoint"]
def __repr__(self) -> str:
return f"<{self.__class__.__name__} login_url={self.login}>"
@dataclass
class Credentials:
"""The Nextcloud Login flow v2 response with app credentials representation."""
def __init__(self, raw_data: dict) -> None:
self.raw_data = raw_data
@property
def server(self) -> str:
"""The address of Nextcloud to connect to.
The server may specify a protocol (http or https). If no protocol is specified https will be used.
"""
return self.raw_data["server"]
@property
def login_name(self) -> str:
"""The username for authenticating with Nextcloud."""
return self.raw_data["loginName"]
@property
def app_password(self) -> str:
"""The application password generated for authenticating with Nextcloud."""
return self.raw_data["appPassword"]
def __repr__(self) -> str:
return f"<{self.__class__.__name__} login={self.login_name} app_password={self.app_password}>"
class _LoginFlowV2API:
"""Class implementing Nextcloud Login flow v2."""
_ep_init: str = "/index.php/login/v2"
_ep_poll: str = "/index.php/login/v2/poll"
def __init__(self, session: NcSession) -> None:
self._session = session
def init(self, user_agent: str = "nc_py_api") -> LoginFlow:
"""Init a Login flow v2.
:param user_agent: Application name. Application password will be associated with this name.
"""
r = self._session.adapter.post(self._ep_init, headers={"user-agent": user_agent})
return LoginFlow(_res_to_json(r))
def poll(self, token: str, timeout: int = MAX_TIMEOUT, step: int = 1, overwrite_auth: bool = True) -> Credentials:
"""Poll the Login flow v2 credentials.
:param token: Token for a polling for confirmation of user authorization.
:param timeout: Maximum time to wait for polling in seconds, defaults to MAX_TIMEOUT.
:param step: Interval for polling in seconds, defaults to 1.
:param overwrite_auth: If True current session will be overwritten with new credentials, defaults to True.
:raises ValueError: If timeout more than 20 minutes.
"""
if timeout > MAX_TIMEOUT:
msg = "Timeout can't be more than 20 minutes."
raise ValueError(msg)
for _ in range(timeout // step):
r = self._session.adapter.post(self._ep_poll, data={"token": token})
if r.status_code == 200:
break
time.sleep(step)
r_model = Credentials(_res_to_json(r))
if overwrite_auth:
self._session.cfg.auth = (r_model.login_name, r_model.app_password)
self._session.init_adapter(restart=True)
self._session.init_adapter_dav(restart=True)
return r_model
class _AsyncLoginFlowV2API:
"""Class implementing Async Nextcloud Login flow v2."""
_ep_init: str = "/index.php/login/v2"
_ep_poll: str = "/index.php/login/v2/poll"
def __init__(self, session: AsyncNcSession) -> None:
self._session = session
async def init(self, user_agent: str = "nc_py_api") -> LoginFlow:
"""Init a Login flow v2.
:param user_agent: Application name. Application password will be associated with this name.
"""
r = await self._session.adapter.post(self._ep_init, headers={"user-agent": user_agent})
return LoginFlow(_res_to_json(r))
async def poll(
self, token: str, timeout: int = MAX_TIMEOUT, step: int = 1, overwrite_auth: bool = True
) -> Credentials:
"""Poll the Login flow v2 credentials.
:param token: Token for a polling for confirmation of user authorization.
:param timeout: Maximum time to wait for polling in seconds, defaults to MAX_TIMEOUT.
:param step: Interval for polling in seconds, defaults to 1.
:param overwrite_auth: If True current session will be overwritten with new credentials, defaults to True.
:raises ValueError: If timeout more than 20 minutes.
"""
if timeout > MAX_TIMEOUT:
raise ValueError("Timeout can't be more than 20 minutes.")
for _ in range(timeout // step):
r = await self._session.adapter.post(self._ep_poll, data={"token": token})
if r.status_code == 200:
break
await asyncio.sleep(step)
r_model = Credentials(_res_to_json(r))
if overwrite_auth:
self._session.cfg.auth = (r_model.login_name, r_model.app_password)
self._session.init_adapter(restart=True)
self._session.init_adapter_dav(restart=True)
return r_model
def _res_to_json(response: httpx.Response) -> dict:
check_error(response)
return json.loads(response.text)
|
