New SDK

.gitignore

@@ -1,2 +1,3 @@
 src/impl/hebece/src/__pycache__
-*.db
+*.db
+__pycache__

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

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"],
+    )

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)

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"

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)

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

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")

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))

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"

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

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

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"],
+        )

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()

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"

sdk/src/apis/prometheus_web/exceptions.py

@@ -0,0 +1,10 @@
+class PrometheusWebException(Exception):
+    pass
+
+
+class InvalidCredentialsException(PrometheusWebException):
+    pass
+
+
+class NoLoggedInException(PrometheusWebException):
+    pass

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

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

sdk/src/interfaces/prometheus/exceptions.py

@@ -0,0 +1,10 @@
+class PrometheusInterfaceException(Exception):
+    pass
+
+
+class NoLoggedInException(PrometheusInterfaceException):
+    pass
+
+
+class NoStudentSelectedException(PrometheusInterfaceException):
+    pass

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)

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

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),
+        )

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"],
+        )

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,
+        )