|
1 | | -import http.cookiejar |
2 | | -import json |
3 | 1 | import logging |
| 2 | +import re |
4 | 3 | import time |
5 | 4 | import traceback |
6 | 5 | from datetime import date |
|
9 | 8 |
|
10 | 9 | from requests import Response, Session |
11 | 10 |
|
12 | | -SESSION_TOKEN_URL = "https://connexion.grdf.fr/api/v1/authn" |
13 | | -SESSION_TOKEN_PAYLOAD = """{{ |
14 | | - "username": "{0}", |
15 | | - "password": "{1}", |
16 | | - "options": {{ |
17 | | - "multiOptionalFactorEnroll": "false", |
18 | | - "warnBeforePasswordExpired": "false" |
19 | | - }} |
| 11 | +START_URL = "https://monespace.grdf.fr/" |
| 12 | + |
| 13 | +MAIL_SESSION_TOKEN_URL = "https://connexion.grdf.fr/idp/idx/identify" |
| 14 | +MAIL_SESSION_TOKEN_PAYLOAD = """{{ |
| 15 | + "identifier": "{0}", |
| 16 | + "stateHandle": "{1}" |
20 | 17 | }}""" |
21 | 18 |
|
22 | | -AUTH_TOKEN_URL = "https://connexion.grdf.fr/login/sessionCookieRedirect" |
23 | | -AUTH_TOKEN_PARAMS = """{{ |
24 | | - "checkAccountSetupComplete": "true", |
25 | | - "token": "{0}", |
26 | | - "redirectUrl": "https://monespace.grdf.fr" |
| 19 | +PASSWORD_SESSION_TOKEN_URL = "https://connexion.grdf.fr/idp/idx/challenge/answer" |
| 20 | +PASSWORD_SESSION_TOKEN_PAYLOAD = """{{ |
| 21 | + "credentials": {{ |
| 22 | + "passcode": "{0}" |
| 23 | + }}, |
| 24 | + "stateHandle": "{1}" |
27 | 25 | }}""" |
28 | 26 |
|
29 | 27 | API_BASE_URL = "https://monespace.grdf.fr/api" |
@@ -71,46 +69,74 @@ def __init__(self, username: str, password: str, retry_count: int = 10): |
71 | 69 | self._username = username |
72 | 70 | self._password = password |
73 | 71 | self._retry_count = retry_count |
74 | | - self._session = None |
| 72 | + self._session: Session | None = None |
75 | 73 |
|
76 | 74 | # ------------------------------------------------------ |
77 | 75 | def login(self): |
78 | 76 | if self._session is not None: |
79 | 77 | return |
80 | 78 |
|
81 | 79 | session = Session() |
82 | | - session.headers.update({"domain": "grdf.fr"}) |
83 | | - session.headers.update({"Content-Type": "application/json"}) |
84 | | - session.headers.update({"X-Requested-With": "XMLHttpRequest"}) |
| 80 | + session.headers.update({"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}) |
85 | 81 |
|
86 | | - payload = SESSION_TOKEN_PAYLOAD.format(self._username, self._password) |
| 82 | + start_response = session.get(START_URL) |
| 83 | + if start_response.status_code != 200: |
| 84 | + raise ServerError( |
| 85 | + f"An error occurred while logging in start. Status code: {start_response.status_code} - {start_response.url}", |
| 86 | + start_response.status_code, |
| 87 | + ) |
87 | 88 |
|
88 | | - response = session.post(SESSION_TOKEN_URL, data=payload) |
| 89 | + pattern = r'"stateToken"\s*:\s*"([^"]+)"' |
| 90 | + match = re.search(pattern, start_response.text) |
| 91 | + if match: |
| 92 | + state_token_html = match.group(1) |
| 93 | + state_token = state_token_html.replace("\\x2D", "-") |
| 94 | + else: |
| 95 | + raise ValueError("Cannot retrieve stateToken inside HTML response") |
| 96 | + |
| 97 | + payload = MAIL_SESSION_TOKEN_PAYLOAD.format(self._username, state_token) |
| 98 | + session.cookies.set("ln", self._username) |
| 99 | + |
| 100 | + mail_response = session.post( |
| 101 | + MAIL_SESSION_TOKEN_URL, |
| 102 | + data=payload, |
| 103 | + headers={"Accept": "application/json; okta-version=1.0.0", "Content-Type": "application/json"}, |
| 104 | + ) |
89 | 105 |
|
90 | | - if response.status_code != 200: |
| 106 | + if mail_response.status_code != 200: |
91 | 107 | raise ServerError( |
92 | | - f"An error occurred while logging in. Status code: {response.status_code} - {response.text}", |
93 | | - response.status_code, |
| 108 | + f"An error occurred while logging in mail. Status code: {mail_response.status_code} - {mail_response.text}", |
| 109 | + mail_response.status_code, |
94 | 110 | ) |
95 | 111 |
|
96 | | - session_token = response.json().get("sessionToken") |
| 112 | + state_handle = mail_response.json().get("stateHandle") |
97 | 113 |
|
98 | | - jar = http.cookiejar.CookieJar() |
| 114 | + payload = PASSWORD_SESSION_TOKEN_PAYLOAD.format(self._password, state_handle) |
99 | 115 |
|
100 | | - self._session = Session() # pylint: disable=attribute-defined-outside-init |
101 | | - self._session.headers.update({"Content-Type": "application/json"}) |
102 | | - self._session.headers.update({"X-Requested-With": "XMLHttpRequest"}) |
| 116 | + password_response = session.post( |
| 117 | + PASSWORD_SESSION_TOKEN_URL, |
| 118 | + data=payload, |
| 119 | + headers={"Accept": "application/json; okta-version=1.0.0", "Content-Type": "application/json"}, |
| 120 | + ) |
103 | 121 |
|
104 | | - params = json.loads(AUTH_TOKEN_PARAMS.format(session_token)) |
| 122 | + if password_response.status_code != 200: |
| 123 | + raise ServerError( |
| 124 | + f"An error occurred while logging in password. Status code: {password_response.status_code} - {password_response.text}", |
| 125 | + password_response.status_code, |
| 126 | + ) |
105 | 127 |
|
106 | | - response = self._session.get(AUTH_TOKEN_URL, params=params, allow_redirects=True, cookies=jar) # type: ignore |
| 128 | + success_url = password_response.json()["success"]["href"] |
107 | 129 |
|
108 | | - if response.status_code != 200: |
| 130 | + response_redirect = session.get(success_url) |
| 131 | + |
| 132 | + if response_redirect.status_code != 200: |
109 | 133 | raise ServerError( |
110 | | - f"An error occurred while getting the auth token. Status code: {response.status_code} - {response.text}", |
111 | | - response.status_code, |
| 134 | + f"An error occurred while logging in response_redirect. Status code: {response_redirect.status_code} - {response_redirect.url}", |
| 135 | + response_redirect.status_code, |
112 | 136 | ) |
113 | 137 |
|
| 138 | + self._session = session |
| 139 | + |
114 | 140 | # ------------------------------------------------------ |
115 | 141 | def is_logged_in(self) -> bool: |
116 | 142 | return self._session is not None |
|
0 commit comments