From 30b7c622969aa4eec2046b48a129950b01f2ef28 Mon Sep 17 00:00:00 2001 From: Marioneq Date: Fri, 7 Feb 2025 21:40:59 +0100 Subject: [PATCH] New SDK --- .gitignore | 3 +- sdk/src/apis/common/models.py | 15 ++ sdk/src/apis/common/utils.py | 12 + sdk/src/apis/efeb/client.py | 56 +++++ sdk/src/apis/efeb/constants.py | 9 + sdk/src/apis/efeb/utils.py | 17 ++ sdk/src/apis/hebe/__init__.py | 2 + sdk/src/apis/hebe/certificate.py | 15 ++ sdk/src/apis/hebe/client.py | 201 +++++++++++++++ sdk/src/apis/hebe/constants.py | 21 ++ sdk/src/apis/hebe/exceptions.py | 42 ++++ sdk/src/apis/hebe/signer.py | 101 ++++++++ sdk/src/apis/hebe/student.py | 58 +++++ sdk/src/apis/prometheus_web/client.py | 109 +++++++++ sdk/src/apis/prometheus_web/constants.py | 11 + sdk/src/apis/prometheus_web/exceptions.py | 10 + sdk/src/interfaces/core/interface.py | 24 ++ sdk/src/interfaces/prometheus/context.py | 47 ++++ sdk/src/interfaces/prometheus/exceptions.py | 10 + sdk/src/interfaces/prometheus/interface.py | 256 ++++++++++++++++++++ sdk/src/interfaces/prometheus/utils.py | 36 +++ sdk/src/models/exam.py | 43 ++++ sdk/src/models/grade.py | 29 +++ sdk/src/models/student.py | 48 ++++ 24 files changed, 1174 insertions(+), 1 deletion(-) create mode 100644 sdk/src/apis/common/models.py create mode 100644 sdk/src/apis/common/utils.py create mode 100644 sdk/src/apis/efeb/client.py create mode 100644 sdk/src/apis/efeb/constants.py create mode 100644 sdk/src/apis/efeb/utils.py create mode 100644 sdk/src/apis/hebe/__init__.py create mode 100644 sdk/src/apis/hebe/certificate.py create mode 100644 sdk/src/apis/hebe/client.py create mode 100644 sdk/src/apis/hebe/constants.py create mode 100644 sdk/src/apis/hebe/exceptions.py create mode 100644 sdk/src/apis/hebe/signer.py create mode 100644 sdk/src/apis/hebe/student.py create mode 100644 sdk/src/apis/prometheus_web/client.py create mode 100644 sdk/src/apis/prometheus_web/constants.py create mode 100644 sdk/src/apis/prometheus_web/exceptions.py create mode 100644 sdk/src/interfaces/core/interface.py create mode 100644 sdk/src/interfaces/prometheus/context.py create mode 100644 sdk/src/interfaces/prometheus/exceptions.py create mode 100644 sdk/src/interfaces/prometheus/interface.py create mode 100644 sdk/src/interfaces/prometheus/utils.py create mode 100644 sdk/src/models/exam.py create mode 100644 sdk/src/models/grade.py create mode 100644 sdk/src/models/student.py diff --git a/.gitignore b/.gitignore index 4433df5..aa644ac 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ src/impl/hebece/src/__pycache__ -*.db \ No newline at end of file +*.db +__pycache__ \ No newline at end of file diff --git a/sdk/src/apis/common/models.py b/sdk/src/apis/common/models.py new file mode 100644 index 0000000..b1ba47f --- /dev/null +++ b/sdk/src/apis/common/models.py @@ -0,0 +1,15 @@ +from dataclasses import dataclass + + +@dataclass +class FsLsQuery: + wa: str + wtrealm: str + wctx: str + + +@dataclass +class FsLsResponse: + wa: str + wresult: str + wctx: str diff --git a/sdk/src/apis/common/utils.py b/sdk/src/apis/common/utils.py new file mode 100644 index 0000000..bdc26c2 --- /dev/null +++ b/sdk/src/apis/common/utils.py @@ -0,0 +1,12 @@ +from bs4 import BeautifulSoup + +from sdk.src.apis.common.models import FsLsResponse + + +def parse_fs_ls_response_form(html: str): + soup = BeautifulSoup(html, "html.parser") + return FsLsResponse( + wa=soup.select_one('input[name="wa"]')["value"], + wresult=soup.select_one('input[name="wresult"]')["value"], + wctx=soup.select_one('input[name="wctx"]')["value"], + ) diff --git a/sdk/src/apis/efeb/client.py b/sdk/src/apis/efeb/client.py new file mode 100644 index 0000000..ca1e2bb --- /dev/null +++ b/sdk/src/apis/efeb/client.py @@ -0,0 +1,56 @@ +import dataclasses +import requests + +from sdk.src.apis.common.models import FsLsResponse, FsLsQuery +from sdk.src.apis.common.utils import parse_fs_ls_response_form +from sdk.src.apis.efeb.constants import ( + ENDPOINT_LOGIN_FS_LS, + ENDPOINT_MESSAGES_APP, + ENDPOINT_STUDENT_APP, + BASE_LOGIN, + BASE_MESSAGES, + BASE_MESSAGES_CE, + BASE_STUDENT, + BASE_STUDENT_CE, +) +from sdk.src.apis.efeb.utils import parse_app_html + + +class EfebClient: + def __init__(self, cookies: dict, symbol: str, is_ce: bool): + self._session = requests.Session() + self._session.cookies.update(cookies) + self._symbol = symbol + self._is_ce = is_ce + + def get_cookies(self): + return self._session.cookies.get_dict() + + def login_fs_ls( + self, query: FsLsQuery, prometheus_response: FsLsResponse | None = None + ): + response = self._session.request( + method="POST" if prometheus_response else "GET", + url=f"{BASE_LOGIN}/{self._symbol}/{ENDPOINT_LOGIN_FS_LS}", + data=( + dataclasses.asdict(prometheus_response) if prometheus_response else None + ), + params=dataclasses.asdict(query), + ) + return parse_fs_ls_response_form(response.text) + + def student_app(self, login_response: FsLsResponse | None = None): + response = self._session.request( + method="POST" if login_response else "GET", + url=f"{BASE_STUDENT_CE if self._is_ce else BASE_STUDENT}/{self._symbol}/{ENDPOINT_STUDENT_APP}", + data=dataclasses.asdict(login_response) if login_response else None, + ) + return parse_app_html(response.text) + + def messages_app(self, login_response: FsLsResponse | None = None): + response = self._session.request( + method="POST" if login_response else "GET", + url=f"{BASE_MESSAGES_CE if self._is_ce else BASE_MESSAGES}/{self._symbol}/{ENDPOINT_MESSAGES_APP}", + data=dataclasses.asdict(login_response) if login_response else None, + ) + return parse_app_html(response.text) diff --git a/sdk/src/apis/efeb/constants.py b/sdk/src/apis/efeb/constants.py new file mode 100644 index 0000000..effb974 --- /dev/null +++ b/sdk/src/apis/efeb/constants.py @@ -0,0 +1,9 @@ +BASE_LOGIN = "https://dziennik-logowanie.vulcan.net.pl" +BASE_STUDENT_CE = "https://uczen.eduvulcan.pl" +BASE_STUDENT = "https://dziennik-uczen.vulcan.net.pl" +BASE_MESSAGES_CE = "https://wiadomosci.eduvulcan.pl" +BASE_MESSAGES = "https://dziennik-wiadomosci.vulcan.net.pl" + +ENDPOINT_LOGIN_FS_LS = "fs/ls" +ENDPOINT_STUDENT_APP = "App" +ENDPOINT_MESSAGES_APP = "App" diff --git a/sdk/src/apis/efeb/utils.py b/sdk/src/apis/efeb/utils.py new file mode 100644 index 0000000..812ebff --- /dev/null +++ b/sdk/src/apis/efeb/utils.py @@ -0,0 +1,17 @@ +from bs4 import BeautifulSoup +import re + + +def parse_fs_ls_form(html: str): + soup = BeautifulSoup(html, "html.parser") + return { + "wa": soup.select_one('input[name="wa"]')["value"], + "wresult": soup.select_one('input[name="wresult"]')["value"], + "wctx": soup.select_one('input[name="wctx"]')["value"], + } + + +def parse_app_html(html: str): + return re.search("appGuid: '(.*?)'", html).group(1), re.search( + "antiForgeryToken: '(.*?)'", html + ).group(1) diff --git a/sdk/src/apis/hebe/__init__.py b/sdk/src/apis/hebe/__init__.py new file mode 100644 index 0000000..db6756d --- /dev/null +++ b/sdk/src/apis/hebe/__init__.py @@ -0,0 +1,2 @@ +from sdk.src.apis.hebe.certificate import Certificate as HebeCertificate +from sdk.src.apis.hebe.client import HebeClient diff --git a/sdk/src/apis/hebe/certificate.py b/sdk/src/apis/hebe/certificate.py new file mode 100644 index 0000000..9ab34ec --- /dev/null +++ b/sdk/src/apis/hebe/certificate.py @@ -0,0 +1,15 @@ +from dataclasses import dataclass +from sdk.src.apis.hebe.signer import generate_key_pair + + +@dataclass +class Certificate: + certificate: str + fingerprint: str + private_key: str + type: str + + @staticmethod + def generate(): + certificate, fingerprint, private_key = generate_key_pair() + return Certificate(certificate, fingerprint, private_key, "X509") diff --git a/sdk/src/apis/hebe/client.py b/sdk/src/apis/hebe/client.py new file mode 100644 index 0000000..6453300 --- /dev/null +++ b/sdk/src/apis/hebe/client.py @@ -0,0 +1,201 @@ +from datetime import date, datetime +import json +import uuid +import requests + +from sdk.src.apis.hebe.certificate import Certificate +from sdk.src.apis.hebe.constants import ( + API_VERSION, + APP_HEBE_NAME, + APP_HEBECE_NAME, + APP_VERSION, + APP_VERSION_CODE, + DEVICE_NAME, + DEVICE_OS, + ENDPOINT_EXAM_BYPUPIL, + ENDPOINT_GRADE_BYPUPIL, + ENDPOINT_HEARTBEAT, + ENDPOINT_REGISTER_HEBE, + ENDPOINT_REGISTER_JWT, + ENDPOINT_REGISTER_TOKEN, + ENDPOINT_SCHOOL_LUCKY, + USER_AGENT, +) +from sdk.src.apis.hebe.exceptions import ( + ExpiredTokenException, + HebeClientException, + InvalidPINException, + InvalidRequestEnvelopeStructure, + InvalidRequestHeadersStructure, + NoPermissionsException, + NoUnitSymbolException, + NotFoundEntityException, + UnauthorizedCertificateException, + UsedTokenException, +) +from sdk.src.apis.hebe.student import HebeStudent +from sdk.src.apis.hebe.signer import get_signature_values +from sdk.src.models.exam import Exam +from sdk.src.models.grade import Grade + + +class HebeClient: + def __init__( + self, certificate: Certificate, rest_url: str | None = None, is_ce: bool = True + ): + self._session = requests.Session() + self._certificate = certificate + self._rest_url = rest_url + self._is_ce = is_ce + + def set_rest_url(self, new_value: str): + self._rest_url = new_value + + def _send_request(self, method: str, endpoint: str, envelope: any = None, **kwargs): + if not self._rest_url: + raise Exception("No rest_url!") + + date = datetime.now() + + url = f"{self._rest_url}/{endpoint}" + body = json.dumps(self._build_body(date, envelope)) if envelope else None + headers = self._build_headers(date, url, body) + response = self._session.request( + method, url, data=body, headers=headers, **kwargs + ) + + data = response.json() + self._check_response_status_code(data["Status"]["Code"]) + + return data["Envelope"] + + def _build_headers(self, date: datetime, url: str, body: any): + digest, canonical_url, signature = get_signature_values( + self._certificate.fingerprint, + self._certificate.private_key, + body, + url, + date, + ) + + headers = { + "signature": signature, + "vcanonicalurl": canonical_url, + "vos": DEVICE_OS, + "vdate": date.strftime("%a, %d %b %Y %H:%M:%S GMT"), + "vapi": API_VERSION, + "vversioncode": APP_VERSION_CODE, + "user-agent": USER_AGENT, + } + if digest: + headers["digest"] = digest + headers["content-type"] = "application/json" + + return headers + + def _build_body(self, date: datetime, envelope: any): + return { + "AppName": APP_HEBECE_NAME if self._is_ce else APP_HEBE_NAME, + "AppVersion": APP_VERSION, + "NotificationToken": "", + "API": int(API_VERSION), + "RequestId": str(uuid.uuid4()), + "Timestamp": int(date.timestamp()), + "TimestampFormatted": date.strftime("%Y-%m-%d %H:%M:%S"), + "Envelope": envelope, + } + + @staticmethod + def _check_response_status_code(status_code: int): + if status_code == 0: + return + if status_code == 100: + raise NoPermissionsException() + if status_code == 101: + raise InvalidRequestEnvelopeStructure() + if status_code == 102: + raise InvalidRequestHeadersStructure() + if status_code == 104: + raise NoUnitSymbolException() + if status_code == 108: + raise UnauthorizedCertificateException() + if status_code == 200: + raise NotFoundEntityException() + if status_code == 201: + raise UsedTokenException() + if status_code == 203: + raise InvalidPINException() + if status_code == 204: + raise ExpiredTokenException() + raise HebeClientException(status_code) + + def register_jwt(self, tokens: list[str]): + envelope = { + "OS": DEVICE_OS, + "Certificate": self._certificate.certificate, + "CertificateType": self._certificate.type, + "DeviceModel": DEVICE_NAME, + "SelfIdentifier": str(uuid.uuid4()), + "CertificateThumbprint": self._certificate.fingerprint, + "Tokens": tokens, + } + return self._send_request("POST", ENDPOINT_REGISTER_JWT, envelope) + + def register_token(self, token: str, pin: str): + # For hebe interface + envelope = { + "OS": DEVICE_OS, + "Certificate": self._certificate.certificate, + "CertificateType": self._certificate.type, + "DeviceModel": DEVICE_NAME, + "SelfIdentifier": str(uuid.uuid4()), + "CertificateThumbprint": self._certificate.fingerprint, + "SecurityToken": token, + "PIN": pin, + } + return self._send_request("POST", ENDPOINT_REGISTER_TOKEN, envelope) + + def get_students(self, mode: int = 2): + envelope = self._send_request( + "GET", ENDPOINT_REGISTER_HEBE, params={"mode": mode} + ) + return map(HebeStudent.from_dict, envelope) + + def heartbeat(self): + self._send_request("GET", ENDPOINT_HEARTBEAT) + + def get_lucky_number(self, student_id: int, constituent_id: int, day: date): + envelope = self._send_request( + "GET", + ENDPOINT_SCHOOL_LUCKY, + params={ + "pupilId": student_id, + "constituentId": constituent_id, + "day": day.strftime("%Y-%m-%d"), + }, + ) + return envelope["Number"] + + def get_grades(self, student_id: int, unit_id: int, period_id: int): + envelope = self._send_request( + "GET", + ENDPOINT_GRADE_BYPUPIL, + params={ + "pupilId": student_id, + "unitId": unit_id, + "periodId": period_id, + }, + ) + return list(map(Grade.from_hebe_dict, envelope)) + + def get_exams(self, student_id: int, from_: date, to: date): + envelope = self._send_request( + "GET", + ENDPOINT_EXAM_BYPUPIL, + params={ + "pupilId": student_id, + "dateFrom": from_, + "dateTo": to, + }, + ) + return list(map(Exam.from_hebe_dict, envelope)) diff --git a/sdk/src/apis/hebe/constants.py b/sdk/src/apis/hebe/constants.py new file mode 100644 index 0000000..89ec08f --- /dev/null +++ b/sdk/src/apis/hebe/constants.py @@ -0,0 +1,21 @@ +DEVICE_OS = "Android" +DEVICE_NAME = "SM-G935F" + +APP_HEBE_NAME = "DzienniczekPlus 2.0" +APP_HEBECE_NAME = "DzienniczekPlus 3.0" +APP_VERSION = "24.11.07 (G)" +APP_VERSION_CODE = "640" + +USER_AGENT = "Dart/3.3 (dart:io)" + +API_VERSION = "1" + +# Endpoints +ENDPOINT_REGISTER_JWT = "mobile/register/jwt" +ENDPOINT_REGISTER_TOKEN = "mobile/register/new" +ENDPOINT_REGISTER_HEBE = "mobile/register/hebe" +ENDPOINT_HEARTBEAT = "mobile/heartbeat" +ENDPOINT_SCHOOL_LUCKY = "mobile/school/lucky" +ENDPOINT_GRADE_BYPUPIL = "mobile/grade/byPupil" +ENDPOINT_SCHEDULE_WITHCHANGES_BYPUPIL = "mobile/schedule/withchanges/byPupil" +ENDPOINT_EXAM_BYPUPIL = "mobile/exam/byPupil" diff --git a/sdk/src/apis/hebe/exceptions.py b/sdk/src/apis/hebe/exceptions.py new file mode 100644 index 0000000..cc5f9fd --- /dev/null +++ b/sdk/src/apis/hebe/exceptions.py @@ -0,0 +1,42 @@ +class HebeClientException(Exception): + pass + + +class NotFoundEndpointException(HebeClientException): + pass + + +class NoUnitSymbolException(HebeClientException): + pass + + +class NoPermissionsException(HebeClientException): + pass + + +class InvalidRequestEnvelopeStructure(HebeClientException): + pass + + +class InvalidRequestHeadersStructure(HebeClientException): + pass + + +class UnauthorizedCertificateException(HebeClientException): + pass + + +class NotFoundEntityException(HebeClientException): + pass + + +class UsedTokenException(HebeClientException): + pass + + +class InvalidPINException(HebeClientException): + pass + + +class ExpiredTokenException(HebeClientException): + pass diff --git a/sdk/src/apis/hebe/signer.py b/sdk/src/apis/hebe/signer.py new file mode 100644 index 0000000..f6f7d93 --- /dev/null +++ b/sdk/src/apis/hebe/signer.py @@ -0,0 +1,101 @@ +import hashlib +import json +import re +import urllib +import base64 +from OpenSSL import crypto +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.asymmetric import padding +from cryptography.hazmat.primitives.serialization import load_der_private_key +from cryptography.hazmat.backends import default_backend + + +def get_encoded_path(full_url): + path = re.search(r"(api/mobile/.+)", full_url) + if path is None: + raise ValueError( + "The URL does not seem correct (does not match `(api/mobile/.+)` regex)" + ) + return urllib.parse.quote(path[1], safe="").lower() + + +def get_digest(body): + if not body: + return None + + m = hashlib.sha256() + m.update(bytes(body, "utf-8")) + return base64.b64encode(m.digest()).decode("utf-8") + + +def get_headers_list(body, digest, canonical_url, timestamp): + sign_data = [ + ["vCanonicalUrl", canonical_url], + ["Digest", digest] if body else None, + ["vDate", timestamp.strftime("%a, %d %b %Y %H:%M:%S GMT")], + ] + + return ( + " ".join(item[0] for item in sign_data if item), + "".join(item[1] for item in sign_data if item), + ) + + +def get_signature(data, private_key): + # Convert data to a string representatio + data_str = json.dumps(data) if isinstance(data, (dict, list)) else str(data) + + # Decode the base64 private key and load it + private_key_bytes = base64.b64decode(private_key) + pkcs8_key = load_der_private_key( + private_key_bytes, password=None, backend=default_backend() + ) + + # Sign the data + signature = pkcs8_key.sign( + bytes(data_str, "utf-8"), padding.PKCS1v15(), hashes.SHA256() + ) + + # Encode the signature in base64 and return + return base64.b64encode(signature).decode("utf-8") + + +def get_signature_values(fingerprint, private_key, body, full_url, timestamp): + canonical_url = get_encoded_path(full_url) + digest = get_digest(body) + headers, values = get_headers_list(body, digest, canonical_url, timestamp) + signature = get_signature(values, private_key) + + return ( + "SHA-256={}".format(digest) if digest else None, + canonical_url, + 'keyId="{}",headers="{}",algorithm="sha256withrsa",signature=Base64(SHA256withRSA({}))'.format( + fingerprint, headers, signature + ), + ) + + +def pem_getraw(pem): + return pem.decode("utf-8").replace("\n", "").split("-----")[2] + + +def generate_key_pair(): + pkcs8 = crypto.PKey() + pkcs8.generate_key(crypto.TYPE_RSA, 2048) + + x509 = crypto.X509() + x509.set_version(2) + x509.set_serial_number(1) + subject = x509.get_subject() + subject.CN = "APP_CERTIFICATE CA Certificate" + x509.set_issuer(subject) + x509.set_pubkey(pkcs8) + x509.sign(pkcs8, "sha256") + x509.gmtime_adj_notBefore(0) + x509.gmtime_adj_notAfter(20 * 365 * 24 * 60 * 60) + + certificate = pem_getraw(crypto.dump_certificate(crypto.FILETYPE_PEM, x509)) + fingerprint = x509.digest("sha1").decode("utf-8").replace(":", "").lower() + private_key = pem_getraw(crypto.dump_privatekey(crypto.FILETYPE_PEM, pkcs8)) + + return certificate, fingerprint, private_key diff --git a/sdk/src/apis/hebe/student.py b/sdk/src/apis/hebe/student.py new file mode 100644 index 0000000..73721d4 --- /dev/null +++ b/sdk/src/apis/hebe/student.py @@ -0,0 +1,58 @@ +from dataclasses import dataclass +from datetime import date + + +@dataclass +class HebePeriod: + id: int + number: int + current: bool + from_: date + to: date + + @staticmethod + def from_dict(data: dict): + return HebePeriod( + id=data["Id"], + number=data["Number"], + current=data["Current"], + from_=date.fromtimestamp(data["Start"]["Timestamp"] / 1000), + to=date.fromtimestamp(data["End"]["Timestamp"] / 1000), + ) + + +@dataclass +class HebeStudent: + id: int + full_name: str + unit_id: int + constituent_id: int + capabilities: list[str] + register_id: int + register_student_number: int | None + class_name: str + is_parent: bool + messagebox_key: str + messagebox_name: str + periods: list[HebePeriod] + rest_url: str + symbol: str + + @staticmethod + def from_dict(data: dict): + return HebeStudent( + id=data["Pupil"]["Id"], + full_name=data["Pupil"]["FirstName"] + " " + data["Pupil"]["Surname"], + unit_id=data["Unit"]["Id"], + constituent_id=data["ConstituentUnit"]["Id"], + capabilities=data["Capabilities"], + register_id=data["Journal"]["Id"], + register_student_number=data["Journal"]["PupilNumber"], + class_name=data["ClassDisplay"], + is_parent=bool(data["CaretakerId"]), + messagebox_key=data["MessageBox"]["GlobalKey"], + messagebox_name=data["MessageBox"]["Name"], + periods=list(map(HebePeriod.from_dict, data["Periods"])), + rest_url=data["Unit"]["RestURL"], + symbol=data["TopLevelPartition"], + ) diff --git a/sdk/src/apis/prometheus_web/client.py b/sdk/src/apis/prometheus_web/client.py new file mode 100644 index 0000000..5bba668 --- /dev/null +++ b/sdk/src/apis/prometheus_web/client.py @@ -0,0 +1,109 @@ +import dataclasses +import json +from bs4 import BeautifulSoup +import requests + +from sdk.src.apis.common.models import FsLsQuery +from sdk.src.apis.common.utils import parse_fs_ls_response_form +from sdk.src.apis.prometheus_web.constants import ( + ENDPOINT_ACCOUNT_QUERY_USER_INFO, + ENDPOINT_API_AP, + ENDPOINT_FS_LS, + ENDPOINT_LOGIN, + HEADERS, + PROMETHEUS_WEB_BASE, +) +from sdk.src.apis.prometheus_web.exceptions import ( + NoLoggedInException, + InvalidCredentialsException, +) + + +class PrometheusWebClient: + def __init__(self, cookies: dict[str, str] | None = None): + self._session = requests.Session() + if cookies: + self._session.cookies.update(cookies) + + def is_logged(self): + if not self._session.cookies.get_dict(): + return False + home_response = self._session.get(PROMETHEUS_WEB_BASE) + soup = BeautifulSoup(home_response.text, "html.parser") + return bool(soup.select_one("div.user-avatar")) + + def get_login_data(self): + login_page = self._session.get( + f"{PROMETHEUS_WEB_BASE}/logowanie", + headers=HEADERS, + ).text + + return self._parse_login_page(login_page) + + def _parse_login_page(self, html: str): + soup = BeautifulSoup(html, "html.parser") + request_verification_token = soup.select_one( + 'input[name="__RequestVerificationToken"]' + )["value"] + + captcha_text = soup.select_one('label[for="captchaUser"]').text + captcha_images = [img["src"] for img in soup.select(".v-captcha-image")] + + return request_verification_token, captcha_text, captcha_images + + def query_user_info(self, username: str, request_verification_token: str): + response = self._session.post( + f"{PROMETHEUS_WEB_BASE}/{ENDPOINT_ACCOUNT_QUERY_USER_INFO}", + { + "alias": username, + "__RequestVerificationToken": request_verification_token, + }, + ) + data = response.json()["data"] + return data["ExtraMessage"], data["ShowCaptcha"] + + def login( + self, + username: str, + password: str, + reqest_verification_token: str, + captcha: str | None, + ): + login_result = self._session.post( + f"{PROMETHEUS_WEB_BASE}/{ENDPOINT_LOGIN}", + data={ + "Alias": username, + "Password": password, + "captchaUser": captcha, + "__RequestVerificationToken": reqest_verification_token, + }, + headers=HEADERS, + ) + + if "Zła nazwa użytkownika lub hasło." in login_result.text: + raise InvalidCredentialsException() + + def get_mobile_data(self): + response = self._session.get( + f"{PROMETHEUS_WEB_BASE}/{ENDPOINT_API_AP}", headers=HEADERS + ) + + if "Logowanie" in response.text: + raise NoLoggedInException() + + soup = BeautifulSoup(response.text, "html.parser") + return json.loads(soup.select_one("input#ap")["value"]) + + def fs_ls(self, query: FsLsQuery): + response = self._session.get( + f"{PROMETHEUS_WEB_BASE}/{ENDPOINT_FS_LS}", + params=dataclasses.asdict(query), + ) + + if "Logowanie" in response.text: + raise NoLoggedInException() + + return parse_fs_ls_response_form(response.text) + + def get_cookies(self): + return self._session.cookies.get_dict() diff --git a/sdk/src/apis/prometheus_web/constants.py b/sdk/src/apis/prometheus_web/constants.py new file mode 100644 index 0000000..0226cb1 --- /dev/null +++ b/sdk/src/apis/prometheus_web/constants.py @@ -0,0 +1,11 @@ +PROMETHEUS_WEB_BASE = "https://eduvulcan.pl" + +HEADERS = { + "User-Agent": "Mozilla/5.0 (Linux; Android 13; SM-G935F Build/TQ3A.230901.001; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/130.0.6723.107 Mobile Safari/537.36", + "X-Requested-With": "pl.edu.vulcan.hebe.ce", +} + +ENDPOINT_LOGIN = "logowanie" +ENDPOINT_API_AP = "api/ap" +ENDPOINT_ACCOUNT_QUERY_USER_INFO = "Account/QueryUserInfo" +ENDPOINT_FS_LS = "fs/ls" diff --git a/sdk/src/apis/prometheus_web/exceptions.py b/sdk/src/apis/prometheus_web/exceptions.py new file mode 100644 index 0000000..19705b7 --- /dev/null +++ b/sdk/src/apis/prometheus_web/exceptions.py @@ -0,0 +1,10 @@ +class PrometheusWebException(Exception): + pass + + +class InvalidCredentialsException(PrometheusWebException): + pass + + +class NoLoggedInException(PrometheusWebException): + pass diff --git a/sdk/src/interfaces/core/interface.py b/sdk/src/interfaces/core/interface.py new file mode 100644 index 0000000..03bdb78 --- /dev/null +++ b/sdk/src/interfaces/core/interface.py @@ -0,0 +1,24 @@ +from datetime import date +from sdk.src.models.exam import Exam +from sdk.src.models.grade import Grade +from sdk.src.models.student import Student + + +class CoreInterface: + def select_student(self, context) -> None: + pass + + def login(self) -> None: + pass + + def get_students(self) -> list[Student]: + pass + + def get_lucky_number(self) -> int: + pass + + def get_grades(period_number: int) -> list[Grade]: + pass + + def get_exams(from_: date, to: date) -> list[Exam]: + pass diff --git a/sdk/src/interfaces/prometheus/context.py b/sdk/src/interfaces/prometheus/context.py new file mode 100644 index 0000000..af40590 --- /dev/null +++ b/sdk/src/interfaces/prometheus/context.py @@ -0,0 +1,47 @@ +from dataclasses import dataclass + +from sdk.src.apis.hebe import HebeCertificate +from sdk.src.apis.hebe.student import HebeStudent +from sdk.src.interfaces.prometheus.utils import get_context_periods_from_hebe_periods + + +@dataclass +class PrometheusStudentContext: + student_id: int + unit_id: int + constituent_id: int + periods: dict[int, int] + register_id: int + hebe_url: str + symbol: str + messagebox_key: str + messagebox_name: str + + @staticmethod + def create(hebe_student: HebeStudent): + return PrometheusStudentContext( + student_id=hebe_student.id, + unit_id=hebe_student.unit_id, + constituent_id=hebe_student.constituent_id, + periods=get_context_periods_from_hebe_periods(hebe_student.periods), + register_id=hebe_student.register_id, + hebe_url=hebe_student.rest_url, + symbol=hebe_student.symbol, + messagebox_key=hebe_student.messagebox_key, + messagebox_name=hebe_student.messagebox_name, + ) + + +@dataclass +class PrometheusWebCredentials: + username: str + password: str + + +@dataclass +class PrometheusAuthContext: + prometheus_web_credentials: PrometheusWebCredentials + symbols: list[str] | None = None + hebe_certificate: HebeCertificate | None = None + prometheus_web_cookies: dict[str, str] | None = None + efeb_web_cookies: dict[str, str | None] | None = None diff --git a/sdk/src/interfaces/prometheus/exceptions.py b/sdk/src/interfaces/prometheus/exceptions.py new file mode 100644 index 0000000..96499f4 --- /dev/null +++ b/sdk/src/interfaces/prometheus/exceptions.py @@ -0,0 +1,10 @@ +class PrometheusInterfaceException(Exception): + pass + + +class NoLoggedInException(PrometheusInterfaceException): + pass + + +class NoStudentSelectedException(PrometheusInterfaceException): + pass diff --git a/sdk/src/interfaces/prometheus/interface.py b/sdk/src/interfaces/prometheus/interface.py new file mode 100644 index 0000000..aa21ccc --- /dev/null +++ b/sdk/src/interfaces/prometheus/interface.py @@ -0,0 +1,256 @@ +from datetime import date, datetime +from sdk.src.apis.common.models import FsLsQuery +from sdk.src.apis.efeb.client import EfebClient +from sdk.src.apis.hebe import HebeClient, HebeCertificate +from sdk.src.apis.prometheus_web.client import PrometheusWebClient +from sdk.src.apis.prometheus_web.exceptions import ( + InvalidCredentialsException, + NoLoggedInException as PrometheusWebNoLoggedInException, +) +from sdk.src.interfaces.core.interface import CoreInterface +from sdk.src.interfaces.prometheus.context import ( + PrometheusStudentContext, + PrometheusAuthContext, +) +from sdk.src.interfaces.prometheus.exceptions import ( + NoLoggedInException, + NoStudentSelectedException, +) +from sdk.src.interfaces.prometheus.utils import ( + flat_map, + get_hebe_url, + parse_jwt_payload, +) +from sdk.src.models.student import Student + + +class PrometheusInterface(CoreInterface): + def __init__( + self, + auth_context: PrometheusAuthContext, + student_context: PrometheusStudentContext | None = None, + ): + self._auth_context = auth_context + self._student_context = student_context + + self._prometheus_web_client = PrometheusWebClient( + self._auth_context.prometheus_web_cookies + ) + self._hebe_client = ( + HebeClient( + certificate=self._auth_context.hebe_certificate, + rest_url=( + self._student_context.hebe_url if self._student_context else None + ), + is_ce=True, + ) + if self._auth_context.hebe_certificate + else None + ) + self._efeb_clients = {} + if self._auth_context.symbols: + for symbol in self._auth_context.symbols: + if ( + self._auth_context.efeb_web_cookies + and self._auth_context.efeb_web_cookies.get(symbol) + ): + self._efeb_clients[symbol] = EfebClient( + cookies=self._auth_context.efeb_web_cookies[symbol], + symbol=symbol, + is_ce=True, + ) + self._efeb_student_vars = {} + self._efeb_messages_vars = {} + + def select_student(self, context: PrometheusStudentContext): + self._student_context = context + self._hebe_client.set_rest_url(context.hebe_url) + self._hebe_client.heartbeat() + + def get_auth_context(self): + return self._auth_context + + def login(self, captcha: str | None = None): + if not self._prometheus_web_client.is_logged(): + result = self._login_prometheus(captcha) + if result: + return result + + self._auth_context.prometheus_web_cookies = ( + self._prometheus_web_client.get_cookies() + ) + + # TODO: Expired hebe certificate + if not self._auth_context.hebe_certificate: + self._login_hebe() + + self._login_efeb() + + def _login_prometheus(self, captcha: str | None = None): + print("Logging for prometheus...") + request_verification_token, captcha_text, captcha_images = ( + self._prometheus_web_client.get_login_data() + ) + + _, show_captcha = self._prometheus_web_client.query_user_info( + self._auth_context.prometheus_web_credentials.username, + request_verification_token, + ) + if show_captcha: + return { + "status": "captcha", + "captcha_text": captcha_text, + "captcha_images": captcha_images, + } + + try: + self._prometheus_web_client.login( + self._auth_context.prometheus_web_credentials.username, + self._auth_context.prometheus_web_credentials.password, + request_verification_token, + captcha, + ) + except InvalidCredentialsException: + _, show_captcha = self._prometheus_web_client.query_user_info( + self._auth_context.prometheus_web_credentials.username, + request_verification_token, + ) + return { + "status": "invalid_credentials", + "captcha_text": captcha_text, + "captcha_images": captcha_images, + "show_captcha": show_captcha, + } + + def _login_hebe(self): + print("Logging for hebe...") + self._auth_context.hebe_certificate = HebeCertificate.generate() + self._hebe_client = HebeClient( + certificate=self._auth_context.hebe_certificate, rest_url=None, is_ce=True + ) + + try: + mobile_data = self._prometheus_web_client.get_mobile_data() + except PrometheusWebNoLoggedInException: + self._auth_context.prometheus_web_cookies = None + raise NoLoggedInException() + + symbols: dict[str, list[any]] = {} + for token in mobile_data["Tokens"]: + payload = parse_jwt_payload(token) + if not symbols.get(payload["tenant"]): + symbols[payload["tenant"]] = [token] + else: + symbols[payload["tenant"]].append(token) + + for symbol in symbols: + self._hebe_client.set_rest_url(get_hebe_url(symbol)) + self._hebe_client.register_jwt(symbols[symbol]) + + self._auth_context.symbols = symbols.keys() + + def _login_efeb(self): + if not self._auth_context.efeb_web_cookies: + self._auth_context.efeb_web_cookies = dict( + [(symbol, None) for symbol in self._auth_context.symbols] + ) + + for symbol in self._auth_context.efeb_web_cookies: + if self._auth_context.efeb_web_cookies[symbol]: + continue + + print(f"Logging for efeb at {symbol}") + + try: + student_prometheus_response = self._prometheus_web_client.fs_ls( + FsLsQuery( + wa="wsignin1.0", + wtrealm=f"https://dziennik-logowanie.vulcan.net.pl/{symbol}/Fs/Ls?wa=wsignin1.0&wtrealm=https://uczen.eduvulcan.pl/{symbol}/App&wctx=auth=studentEV&nslo=1", + wctx="nslo=1", + ) + ) + except PrometheusWebNoLoggedInException: + self._auth_context.prometheus_web_cookies = None + raise NoLoggedInException() + + self._efeb_clients[symbol] = EfebClient( + cookies=self._auth_context.prometheus_web_cookies, + symbol=symbol, + is_ce=True, + ) + student_login_response = self._efeb_clients[symbol].login_fs_ls( + query=FsLsQuery( + wa="wsignin1.0", + wtrealm=f"https://uczen.eduvulcan.pl/{symbol}/App", + wctx="auth=studentEV&nslo=1", + ), + prometheus_response=student_prometheus_response, + ) + self._efeb_student_vars[symbol] = self._efeb_clients[symbol].student_app( + login_response=student_login_response + ) + + messages_login_response = self._efeb_clients[symbol].login_fs_ls( + query=FsLsQuery( + wa="wsignin1.0", + wtrealm=f"https://wiadomosci.eduvulcan.pl/{symbol}/App", + wctx="auth=studentEV&nslo=1", + ), + ) + self._efeb_messages_vars[symbol] = self._efeb_clients[symbol].messages_app( + login_response=messages_login_response + ) + self._auth_context.efeb_web_cookies[symbol] = self._efeb_clients[ + symbol + ].get_cookies() + + def _check_is_auth_context_full(self): + if ( + not self._auth_context.hebe_certificate + or not self._auth_context.prometheus_web_cookies + or not self._auth_context.symbols + ): + raise NoLoggedInException() + + def _check_is_student_selected(self): + if not self._student_context: + raise NoStudentSelectedException() + + def get_students(self): + self._check_is_auth_context_full() + return flat_map( + [*map(self._get_students_in_symbol, self._auth_context.symbols)] + ) + + def _get_students_in_symbol(self, symbol: str): + self._hebe_client.set_rest_url(get_hebe_url(symbol)) + hebe_students = self._hebe_client.get_students() + return [ + Student.from_hebe_student( + hebe_student, PrometheusStudentContext.create(hebe_student) + ) + for hebe_student in hebe_students + ] + + def get_lucky_number(self): + self._check_is_auth_context_full() + self._check_is_student_selected() + return self._hebe_client.get_lucky_number( + self._student_context.student_id, + self._student_context.constituent_id, + datetime.now(), + ) + + def get_grades(self, period_number: int): + self._check_is_auth_context_full() + self._check_is_student_selected() + return self._hebe_client.get_grades( + self._student_context.student_id, + self._student_context.unit_id, + self._student_context.periods[period_number], + ) + + def get_exams(self, from_: date, to: date): + self._check_is_auth_context_full() + self._check_is_student_selected() + return self._hebe_client.get_exams(self._student_context.student_id, from_, to) diff --git a/sdk/src/interfaces/prometheus/utils.py b/sdk/src/interfaces/prometheus/utils.py new file mode 100644 index 0000000..ade7b2d --- /dev/null +++ b/sdk/src/interfaces/prometheus/utils.py @@ -0,0 +1,36 @@ +import base64 +import json + +from sdk.src.apis.hebe.student import HebePeriod + + +def parse_jwt_payload(token: str): + # Split the JWT into parts + _, payload, _ = token.split(".") + + # Decode the payload from Base64 + # Add padding + payload += "=" * (-len(payload) % 4) + decoded_payload = base64.urlsafe_b64decode(payload).decode("utf-8") + + # Parse the payload as JSON + payload_json = json.loads(decoded_payload) + return payload_json + + +def get_hebe_url(symbol: str, unit_symbol: str | None = None): + return f"https://lekcjaplus.vulcan.net.pl/{symbol}{'/' + unit_symbol if unit_symbol else ''}/api" + + +def get_context_periods_from_hebe_periods(hebe_periods: list[HebePeriod]): + periods = {} + for period in hebe_periods: + periods[period.number] = period.id + return periods + + +def flat_map(list: list[list]): + new = [] + for sublist in list: + [new.append(item) for item in sublist] + return new diff --git a/sdk/src/models/exam.py b/sdk/src/models/exam.py new file mode 100644 index 0000000..e892aa2 --- /dev/null +++ b/sdk/src/models/exam.py @@ -0,0 +1,43 @@ +from dataclasses import dataclass +from datetime import date, datetime +from enum import Enum + + +class ExamType(Enum): + TEST = 0 + SHORT_TEST = 1 + CLASSWORK = 2 + OTHER = 3 + + @staticmethod + def from_hebe_type_name(type_name: str): + match type_name: + case "Sprawdzian": + return ExamType.TEST + case "Kartkówka": + return ExamType.SHORT_TEST + case "Praca klasowa": + return ExamType.CLASSWORK + case _: + return ExamType.OTHER + + +@dataclass +class Exam: + deadline: date + subject: str + type: ExamType + description: str + creator: str + created_at: datetime + + @staticmethod + def from_hebe_dict(data: dict): + return Exam( + deadline=datetime.fromtimestamp(data["Deadline"]["Timestamp"] / 1000), + subject=data["Subject"]["Name"], + type=ExamType.from_hebe_type_name(data["Type"]), + description=data["Content"], + creator=data["Creator"]["DisplayName"], + created_at=datetime.fromtimestamp(data["DateCreated"]["Timestamp"] / 1000), + ) diff --git a/sdk/src/models/grade.py b/sdk/src/models/grade.py new file mode 100644 index 0000000..8f5e144 --- /dev/null +++ b/sdk/src/models/grade.py @@ -0,0 +1,29 @@ +from dataclasses import dataclass +from datetime import datetime + + +@dataclass +class Grade: + value: str + is_point: bool + point_numerator: int + point_denominator: int + weight: float + name: str + created_at: datetime + subject: str + creator: str + + @staticmethod + def from_hebe_dict(data: dict[str, any]): + return Grade( + value=data["Content"], + is_point=data["Numerator"] != None, + point_numerator=data["Numerator"], + point_denominator=data["Denominator"], + weight=data["Column"]["Weight"], + name=data["Column"]["Name"] or data["Column"]["Code"], + created_at=datetime.fromtimestamp(data["DateCreated"]["Timestamp"] / 1000), + subject=data["Column"]["Subject"]["Name"], + creator=data["Creator"]["DisplayName"], + ) diff --git a/sdk/src/models/student.py b/sdk/src/models/student.py new file mode 100644 index 0000000..e5e26cb --- /dev/null +++ b/sdk/src/models/student.py @@ -0,0 +1,48 @@ +from dataclasses import dataclass +from datetime import date +from typing import Generic, TypeVar + +from sdk.src.apis.hebe.student import HebePeriod, HebeStudent + + +@dataclass +class Period: + id: int + number: int + current: bool + from_: date + to: date + + @staticmethod + def from_hebe_period(period: HebePeriod): + return Period( + id=period.id, + number=period.number, + current=period.current, + from_=period.from_, + to=period.to, + ) + + +T = TypeVar("T") + + +@dataclass +class Student(Generic[T]): + full_name: str + is_parent: bool + class_name: str + register_number: int | None + periods: list[Period] + context: T + + @staticmethod + def from_hebe_student(student: HebeStudent, context: T): + return Student( + full_name=student.full_name, + is_parent=student.is_parent, + class_name=student.class_name, + register_number=student.register_student_number, + periods=list(map(Period.from_hebe_period, student.periods)), + context=context, + )