tgapi.py

from typing import Optional, Tuple, Any
import re
import requests
from urllib.parse import quote
from lxml import html
from fake_useragent import FakeUserAgent

fake_useragent = FakeUserAgent()

class TelegramApplication:
    def __init__(
        self,
        phone_number: str,
        app_title: str = "",
        app_shortname: str = "",
        app_url: str = "",
        app_platform: str = "desktop",
        app_desc: str = "",
        random_hash: Optional[str] = None,
        stel_token: Optional[str] = None,
        useragent: Optional[str] = None
    ) -> None:
        # Sanitize and validate phone number
        if not self._validate_phone_number(phone_number):
            raise ValueError("Invalid phone number format")
        self.phone_number = phone_number.strip()

        # Sanitize app parameters
        self.app_title = self._sanitize_input(app_title, max_length=255)
        self.app_shortname = self._sanitize_input(app_shortname, max_length=64)
        self.app_url = self._sanitize_url(app_url)
        self.app_platform = app_platform.lower() if app_platform in {"desktop", "ios", "android"} else "desktop"
        self.app_desc = self._sanitize_input(app_desc, max_length=255)

        # Secure sensitive fields
        self._random_hash = random_hash
        self._stel_token = stel_token
        self.useragent = useragent or fake_useragent.random

    def _validate_phone_number(self, phone: str) -> bool:
        """Validate phone number format."""
        phone_pattern = r'^\+?[1-9]\d{1,14}$'
        return bool(re.match(phone_pattern, phone.strip()))

    def _sanitize_input(self, value: str, max_length: int) -> str:
        """Sanitize input strings."""
        if not value:
            return ""
        # Remove dangerous characters and limit length
        sanitized = re.sub(r'[^\w\s\-\.]', '', value.strip())
        return sanitized[:max_length]

    def _sanitize_url(self, url: str) -> str:
        """Sanitize and validate URL."""
        if not url:
            return ""
        # Basic URL validation
        url_pattern = r'^https?://[\w\-\.]+(?:/[\w\-\./]*)*$'
        sanitized = self._sanitize_input(url, max_length=2048)
        return sanitized if re.match(url_pattern, sanitized) else ""

    def _get_headers(self, is_post: bool = True, referer: str = "https://my.telegram.org/auth") -> dict:
        """Generate common headers for requests."""
        headers = {
            "Origin": "https://my.telegram.org",
            "Accept-Encoding": "gzip, deflate, br",
            "Accept-Language": "en-US,en;q=0.9",
            "User-Agent": self.useragent,
            "Accept": "application/json, text/javascript, */*; q=0.01" if is_post else "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
            "Referer": referer,
            "Connection": "keep-alive",
            "DNT": "1"
        }
        if is_post:
            headers["Content-Type"] = "application/x-www-form-urlencoded; charset=UTF-8"
            headers["X-Requested-With"] = "XMLHttpRequest"
        return headers

    def send_password(self) -> bool:
        """Send password request to Telegram."""
        try:
            response = requests.post(
                url="https://my.telegram.org/auth/send_password",
                data=f"phone={quote(self.phone_number)}",
                headers=self._get_headers(),
                timeout=10
            )
            response.raise_for_status()
            data = response.json()
            if "random_hash" not in data:
                return False
            self._random_hash = data["random_hash"]
            return bool(self._random_hash)
        except (requests.RequestException, ValueError):
            return False

    def auth_login(self, cloud_password: str) -> bool:
        """Authenticate with cloud password."""
        if not self._random_hash or not cloud_password:
            return False
        try:
            sanitized_password = self._sanitize_input(cloud_password, max_length=100)
            response = requests.post(
                url="https://my.telegram.org/auth/login",
                data=f"phone={quote(self.phone_number)}&random_hash={quote(self._random_hash)}&password={quote(sanitized_password)}",
                headers=self._get_headers(),
                timeout=10
            )
            response.raise_for_status()
            self._stel_token = response.cookies.get("stel_token")
            return bool(self._stel_token)
        except (requests.RequestException, KeyError):
            return False

    def auth_app(self) -> Optional[Tuple[str, str]]:
        """Retrieve or create Telegram app credentials."""
        if not self._stel_token:
            return None
        try:
            headers = self._get_headers(is_post=False, referer="https://my.telegram.org/org")
            headers["Cookie"] = f"stel_token={quote(self._stel_token)}"
            headers["Upgrade-Insecure-Requests"] = "1"
            headers["Cache-Control"] = "max-age=0"

            response = requests.get(
                url="https://my.telegram.org/apps",
                headers=headers,
                timeout=10
            )
            response.raise_for_status()
            tree = html.fromstring(response.content)
            api_fields = tree.xpath('//span[@class="form-control input-xlarge uneditable-input"]//text()')
            if len(api_fields) >= 2:
                return api_fields[0], api_fields[1]

            # Try to create a new app
            hidden_hash = tree.xpath('//input[@name="hash"]/@value')
            if not hidden_hash:
                return None

            requests.post(
                url="https://my.telegram.org/apps/create",
                data=f"hash={quote(hidden_hash[0])}&app_title={quote(self.app_title)}&app_shortname={quote(self.app_shortname)}&app_url={quote(self.app_url)}&app_platform={self.app_platform}&app_desc={quote(self.app_desc)}",
                headers=self._get_headers(referer="https://my.telegram.org/apps"),
                timeout=10
            ).raise_for_status()

            response = requests.get(
                url="https://my.telegram.org/apps",
                headers=headers,
                timeout=10
            )
            response.raise_for_status()
            tree = html.fromstring(response.content)
            api_fields = tree.xpath('//span[@class="form-control input-xlarge uneditable-input"]//text()')
            return (api_fields[0], api_fields[1]) if len(api_fields) >= 2 else None
        except (requests.RequestException, IndexError, ValueError):
            return None

    def __setattr__(self, name: str, value: Any) -> None:
        """Secure sensitive attributes."""
        if name in ("_random_hash", "_stel_token") and value is not None:
            object.__setattr__(self, name, value[:100])  # Limit length
        else:
            object.__setattr__(self, name, value)