集成CookieCloud服务端

.gitignore

@@ -10,7 +10,9 @@ app/helper/*.pyd
 app/helper/*.bin
 app/plugins/**
 !app/plugins/__init__.py
+config/cookies/**
 config/user.db
 config/sites/**
 *.pyc
 *.log
+.vscode

app/api/servcookie.py

@@ -0,0 +1,104 @@
+import gzip
+import json
+import os
+from typing import Annotated, Any, Callable, Dict
+
+from fastapi import APIRouter, Depends, HTTPException, Path, Request, Response
+from fastapi.responses import PlainTextResponse
+from fastapi.routing import APIRoute
+
+from app import schemas
+from app.core.config import settings
+from app.utils.common import get_decrypted_cookie_data
+
+
+class GzipRequest(Request):
+
+    async def body(self) -> bytes:
+        if not hasattr(self, "_body"):
+            body = await super().body()
+            if "gzip" in self.headers.getlist("Content-Encoding"):
+                body = gzip.decompress(body)
+            self._body = body
+        return self._body
+
+
+class GzipRoute(APIRoute):
+
+    def get_route_handler(self) -> Callable:
+        original_route_handler = super().get_route_handler()
+
+        async def custom_route_handler(request: Request) -> Response:
+            request = GzipRequest(request.scope, request.receive)
+            return await original_route_handler(request)
+
+        return custom_route_handler
+
+
+async def verify_server_enabled():
+    """
+    校验CookieCloud服务路由是否打开
+    """
+    if not settings.COOKIECLOUD_ENABLE_LOCAL:
+        raise HTTPException(status_code=400, detail="本地CookieCloud服务器未启用")
+    return True
+
+
+cookie_router = APIRouter(route_class=GzipRoute,
+                          tags=['servcookie'],
+                          dependencies=[Depends(verify_server_enabled)])
+
+
+@cookie_router.get("/", response_class=PlainTextResponse)
+def get_root():
+    return "Hello World! API ROOT = /cookiecloud"
+
+
+@cookie_router.post("/", response_class=PlainTextResponse)
+def post_root():
+    return "Hello World! API ROOT = /cookiecloud"
+
+
+@cookie_router.post("/update")
+async def update_cookie(req: schemas.CookieData):
+    file_path = os.path.join(settings.COOKIE_PATH,
+                             os.path.basename(req.uuid) + ".json")
+    content = json.dumps({"encrypted": req.encrypted})
+    with open(file_path, encoding="utf-8", mode="w") as file:
+        file.write(content)
+    read_content = None
+    with open(file_path, encoding="utf-8", mode="r") as file:
+        read_content = file.read()
+    if (read_content == content):
+        return {"action": "done"}
+    else:
+        return {"action": "error"}
+
+
+def load_encrypt_data(uuid: str) -> Dict[str, Any]:
+    file_path = os.path.join(settings.COOKIE_PATH,
+                             os.path.basename(uuid) + ".json")
+
+    # 检查文件是否存在
+    if not os.path.exists(file_path):
+        raise HTTPException(status_code=404, detail="Item not found")
+
+    # 读取文件
+    with open(file_path, encoding="utf-8", mode="r") as file:
+        read_content = file.read()
+    data = json.loads(read_content)
+    return data
+
+
+@cookie_router.get("/get/{uuid}")
+async def get_cookie(
+        uuid: Annotated[str, Path(min_length=5, pattern="^[a-zA-Z0-9]+$")]):
+    return load_encrypt_data(uuid)
+
+
+@cookie_router.post("/get/{uuid}")
+async def post_cookie(
+        uuid: Annotated[str, Path(min_length=5, pattern="^[a-zA-Z0-9]+$")],
+        request: schemas.CookiePassword):
+    data = load_encrypt_data(uuid)
+    return get_decrypted_cookie_data(uuid, request.password, data["encrypted"])

app/chain/site.py

@@ -43,7 +43,9 @@ class SiteChain(ChainBase):
         self.cookiecloud = CookieCloudHelper(
             server=settings.COOKIECLOUD_HOST,
             key=settings.COOKIECLOUD_KEY,
-            password=settings.COOKIECLOUD_PASSWORD
+            password=settings.COOKIECLOUD_PASSWORD,
+            enable_local=settings.COOKIECLOUD_ENABLE_LOCAL,
+            local_path=settings.COOKIE_PATH
         )
 
         # 特殊站点登录验证

app/core/config.py

@@ -187,6 +187,8 @@ class Settings(BaseSettings):
     PLEX_TOKEN: Optional[str] = None
     # 转移方式 link/copy/move/softlink
     TRANSFER_TYPE: str = "copy"
+    # CookieCloud是否启动本地服务
+    COOKIECLOUD_ENABLE_LOCAL: Optional[bool] = True
     # CookieCloud服务器地址
     COOKIECLOUD_HOST: str = "https://movie-pilot.org/cookiecloud"
     # CookieCloud用户KEY
@@ -275,6 +277,10 @@ class Settings(BaseSettings):
     @property
     def LOG_PATH(self):
         return self.CONFIG_PATH / "logs"
+    
+    @property
+    def COOKIE_PATH(self):
+        return self.CONFIG_PATH / "cookies"
 
     @property
     def CACHE_CONF(self):
@@ -397,6 +403,9 @@ class Settings(BaseSettings):
         with self.LOG_PATH as p:
             if not p.exists():
                 p.mkdir(parents=True, exist_ok=True)
+        with self.COOKIE_PATH as p:
+            if not p.exists():
+                p.mkdir(parents=True, exist_ok=True)
 
     class Config:
         case_sensitive = True

app/helper/cookiecloud.py

@@ -1,6 +1,7 @@
+import os
 import json
 
-from typing import Tuple, Optional
+from typing import Any, Dict, Tuple, Optional
 from hashlib import md5
 
 from app.utils.http import RequestUtils
@@ -12,10 +13,12 @@ class CookieCloudHelper:
 
     _ignore_cookies: list = ["CookieAutoDeleteBrowsingDataCleanup", "CookieAutoDeleteCleaningDiscarded"]
 
-    def __init__(self, server: str, key: str, password: str):
+    def __init__(self, server: str, key: str, password: str, enable_local: bool, local_path: str):
         self._server = server
         self._key = key
         self._password = password
+        self._enable_local = enable_local
+        self._local_path = local_path
         self._req = RequestUtils(content_type="application/json")
 
     def download(self) -> Tuple[Optional[dict], str]:
@@ -23,68 +26,78 @@ class CookieCloudHelper:
         从CookieCloud下载数据
         :return: Cookie数据、错误信息
         """
-        if not self._server or not self._key or not self._password:
+        if (not self._server and
+                not self._enable_local) or not self._key or not self._password:
             return None, "CookieCloud参数不正确"
-        req_url = "%s/get/%s" % (self._server, str(self._key).strip())
-        ret = self._req.get_res(url=req_url)
-        if ret and ret.status_code == 200:
-            result = ret.json()
-            if not result:
-                return {}, "未下载到数据"        
-            encrypted = result.get("encrypted")
-            if not encrypted:
-                return {}, "未获取到cookie密文"
-            else:
-                crypt_key = self.get_crypt_key()
-                try:
-                    decrypted_data = decrypt(encrypted, crypt_key).decode('utf-8')
-                    result = json.loads(decrypted_data)
-                except Exception as e:
-                    return {}, "cookie解密失败" + str(e)
 
+        result = None
+        if self._enable_local:
+            # 开启本地服务时，从本地直接读取数据
+            result = self.load_local_encrypt_data(self._key)
             if not result:
-                return {}, "cookie解密为空"
+                return {}, "未从本地CookieCloud服务加载到cookie数据"
-            
-            if result.get("cookie_data"):
-                contents = result.get("cookie_data")
-            else:
-                contents = result
-            # 整理数据,使用domain域名的最后两级作为分组依据
-            domain_groups = {}
-            for site, cookies in contents.items():
-                for cookie in cookies:
-                    domain_key = StringUtils.get_url_domain(cookie.get("domain"))
-                    if not domain_groups.get(domain_key):
-                        domain_groups[domain_key] = [cookie]
-                    else:
-                        domain_groups[domain_key].append(cookie)
-            # 返回错误
-            ret_cookies = {}
-            # 索引器
-            for domain, content_list in domain_groups.items():
-                if not content_list:
-                    continue
-                # 只有cf的cookie过滤掉
-                cloudflare_cookie = True
-                for content in content_list:
-                    if content["name"] != "cf_clearance":
-                        cloudflare_cookie = False
-                        break
-                if cloudflare_cookie:
-                    continue
-                # 站点Cookie
-                cookie_str = ";".join(
-                    [f"{content.get('name')}={content.get('value')}"
-                     for content in content_list
-                     if content.get("name") and content.get("name") not in self._ignore_cookies]
-                )
-                ret_cookies[domain] = cookie_str
-            return ret_cookies, ""
-        elif ret:
-            return None, f"同步CookieCloud失败，错误码：{ret.status_code}"
         else:
-            return None, "CookieCloud请求失败，请检查服务器地址、用户KEY及加密密码是否正确"
+            req_url = "%s/get/%s" % (self._server, str(self._key).strip())
-        
+            ret = self._req.get_res(url=req_url)
+            if ret and ret.status_code == 200:
+                result = ret.json()
+                if not result:
+                    return {}, "未从" + self._server + "下载到数据"
+            elif ret:
+                return None, f"远程同步CookieCloud失败，错误码：{ret.status_code}"
+            else:
+                return None, "CookieCloud请求失败，请检查服务器地址、用户KEY及加密密码是否正确"
+
+        encrypted = result.get("encrypted")
+        if not encrypted:
+            return {}, "未获取到cookie密文"
+        else:
+            crypt_key = self.get_crypt_key()
+            try:
+                decrypted_data = decrypt(encrypted, crypt_key).decode('utf-8')
+                result = json.loads(decrypted_data)
+            except Exception as e:
+                return {}, "cookie解密失败" + str(e)
+
+        if not result:
+            return {}, "cookie解密为空"
+
+        if result.get("cookie_data"):
+            contents = result.get("cookie_data")
+        else:
+            contents = result
+        # 整理数据,使用domain域名的最后两级作为分组依据
+        domain_groups = {}
+        for site, cookies in contents.items():
+            for cookie in cookies:
+                domain_key = StringUtils.get_url_domain(cookie.get("domain"))
+                if not domain_groups.get(domain_key):
+                    domain_groups[domain_key] = [cookie]
+                else:
+                    domain_groups[domain_key].append(cookie)
+        # 返回错误
+        ret_cookies = {}
+        # 索引器
+        for domain, content_list in domain_groups.items():
+            if not content_list:
+                continue
+            # 只有cf的cookie过滤掉
+            cloudflare_cookie = True
+            for content in content_list:
+                if content["name"] != "cf_clearance":
+                    cloudflare_cookie = False
+                    break
+            if cloudflare_cookie:
+                continue
+            # 站点Cookie
+            cookie_str = ";".join(
+                [f"{content.get('name')}={content.get('value')}"
+                    for content in content_list
+                    if content.get("name") and content.get("name") not in self._ignore_cookies]
+            )
+            ret_cookies[domain] = cookie_str
+        return ret_cookies, ""
+
     def get_crypt_key(self) -> bytes:
         """
         使用UUID和密码生成CookieCloud的加解密密钥
@@ -92,3 +105,16 @@ class CookieCloudHelper:
         md5_generator = md5()
         md5_generator.update((str(self._key).strip() + '-' + str(self._password).strip()).encode('utf-8'))
         return (md5_generator.hexdigest()[:16]).encode('utf-8')
+
+    def load_local_encrypt_data(self,uuid: str) -> Dict[str, Any]:
+        file_path = os.path.join(self._local_path, os.path.basename(uuid) + ".json")
+
+        # 检查文件是否存在
+        if not os.path.exists(file_path):
+            return None
+
+        # 读取文件
+        with open(file_path, encoding="utf-8", mode="r") as file:
+            read_content = file.read()
+        data = json.loads(read_content)
+        return data

app/main.py

@@ -53,10 +53,13 @@ def init_routers():
     """
     from app.api.apiv1 import api_router
     from app.api.servarr import arr_router
+    from app.api.servcookie import cookie_router
     # API路由
     App.include_router(api_router, prefix=settings.API_V1_STR)
     # Radarr、Sonarr路由
     App.include_router(arr_router, prefix="/api/v3")
+    # CookieCloud路由
+    App.include_router(cookie_router, prefix="/cookiecloud")
 
 
 def start_frontend():

app/schemas/__init__.py

@@ -5,6 +5,7 @@ from .site import *
 from .subscribe import *
 from .context import *
 from .servarr import *
+from .servcookie import *
 from .plugin import *
 from .history import *
 from .dashboard import *

app/schemas/servcookie.py

@@ -0,0 +1,13 @@
+from typing import Union
+
+from fastapi import Query
+from pydantic import BaseModel
+
+
+class CookieData(BaseModel):
+    uuid: str = Query(min_length=5, pattern="^[a-zA-Z0-9]+$")
+    encrypted: str = Query(min_length=1, max_length=1024 * 1024 * 50)
+
+
+class CookiePassword(BaseModel):
+    password: str

app/utils/common.py

@@ -1,10 +1,12 @@
-import time
 import base64
+import json
+import time
+
+from hashlib import md5
+from typing import Any, Dict, Optional
 
-from typing import Any
 from Crypto import Random
 from Crypto.Cipher import AES
-from hashlib import md5
 
 
 def retry(ExceptionToCheck: Any,
@@ -82,3 +84,21 @@ def decrypt(encrypted: str | bytes, passphrase: bytes) -> bytes:
     aes = AES.new(key, AES.MODE_CBC, iv)
     data = aes.decrypt(encrypted[16:])
     return data[:-(data[-1] if type(data[-1]) == int else ord(data[-1]))]
+
+
+def get_decrypted_cookie_data(uuid: str, password: str,
+                              encrypted: str) -> Optional[Dict[str, Any]]:
+    key_md5 = md5()
+    key_md5.update((uuid + '-' + password).encode('utf-8'))
+    aes_key = (key_md5.hexdigest()[:16]).encode('utf-8')
+
+    if encrypted is not None:
+        try:
+            decrypted_data = decrypt(encrypted, aes_key).decode('utf-8')
+            decrypted_data = json.loads(decrypted_data)
+            if 'cookie_data' in decrypted_data:
+                return decrypted_data
+        except Exception as e:
+            return None
+    else:
+        return None