majuna/app/models/base.py

from datetime import datetime
from typing import List

from app.brm.brn import BRN
from app.extensions import db
from app.models import AbstractConfiguration


class Group(AbstractConfiguration):
    group_name = db.Column(db.String(80), unique=True, nullable=False)
    eotk = db.Column(db.Boolean())

    origins = db.relationship("Origin", back_populates="group")
    bridgeconfs = db.relationship("BridgeConf", back_populates="group")
    eotks = db.relationship("Eotk", back_populates="group")
    onions = db.relationship("Onion", back_populates="group")
    smart_proxies = db.relationship("SmartProxy", back_populates="group")
    pools = db.relationship("Pool", secondary="pool_group", back_populates="groups")

    @classmethod
    def csv_header(cls) -> List[str]:
        return super().csv_header() + [
            "group_name", "eotk"
        ]

    @property
    def brn(self) -> BRN:
        return BRN(
            group_id=self.id,
            product="group",
            provider="",
            resource_type="group",
            resource_id=str(self.id)
        )


class Pool(AbstractConfiguration):
    pool_name = db.Column(db.String(80), unique=True, nullable=False)
    api_key = db.Column(db.String(80), nullable=False)
    redirector_domain = db.Column(db.String(128), nullable=True)

    @classmethod
    def csv_header(cls) -> List[str]:
        return super().csv_header() + [
            "pool_name"
        ]

    proxies = db.relationship("Proxy", back_populates="pool")
    lists = db.relationship("MirrorList", back_populates="pool")
    groups = db.relationship("Group", secondary="pool_group", back_populates="pools")

    @property
    def brn(self) -> BRN:
        return BRN(
            group_id=0,
            product="pool",
            provider="",
            resource_type="pool",
            resource_id=str(self.pool_name)
        )


class PoolGroup(db.Model):  # type: ignore[name-defined,misc]
    pool_id = db.Column(db.Integer, db.ForeignKey("pool.id"), primary_key=True)
    group_id = db.Column(db.Integer, db.ForeignKey("group.id"), primary_key=True)


class MirrorList(AbstractConfiguration):
    pool_id = db.Column(db.Integer, db.ForeignKey("pool.id"))
    provider = db.Column(db.String(255), nullable=False)
    format = db.Column(db.String(20), nullable=False)
    encoding = db.Column(db.String(20), nullable=False)
    container = db.Column(db.String(255), nullable=False)
    branch = db.Column(db.String(255), nullable=False)
    role = db.Column(db.String(255), nullable=True)
    filename = db.Column(db.String(255), nullable=False)

    pool = db.relationship("Pool", back_populates="lists")

    providers_supported = {
        "github": "GitHub",
        "gitlab": "GitLab",
        "http_post": "HTTP POST",
        "s3": "AWS S3",
    }

    formats_supported = {
        "bc2": "Bypass Censorship v2",
        "bc3": "Bypass Censorship v3",
        "bca": "Bypass Censorship Analytics",
        "bridgelines": "Tor Bridge Lines",
        "rdr": "Redirector Data"
    }

    encodings_supported = {
        "json": "JSON (Plain)",
        "jsno": "JSON (Obfuscated)",
        "js": "JavaScript (Plain)",
        "jso": "JavaScript (Obfuscated)"
    }

    def destroy(self) -> None:
        self.destroyed = datetime.utcnow()
        self.updated = datetime.utcnow()

    def url(self) -> str:
        if self.provider == "gitlab":
            return f"https://gitlab.com/{self.container}/-/raw/{self.branch}/{self.filename}"
        if self.provider == "github":
            return f"https://raw.githubusercontent.com/{self.container}/{self.branch}/{self.filename}"
        if self.provider == "s3":
            return f"s3://{self.container}/{self.filename}"
        if self.provider == "http_post":
            return str(self.container)
        return "Unknown provider"

    @classmethod
    def csv_header(cls) -> List[str]:
        return super().csv_header() + [
            "provider", "format", "container", "branch", "filename"
        ]

    @property
    def brn(self) -> BRN:
        return BRN(
            group_id=0,
            product="list",
            provider=self.provider,
            resource_type="list",
            resource_id=str(self.id)
        )