cms-SiteForge/app/core.py

"""Shared site/route/content rules, with content sanitization."""
import ipaddress
import json
import re
import socket
import uuid
from urllib.parse import urlparse, unquote
import bleach
from . import db

TYPES = ('page','product','post','service','case','brand','vehicle','other')
ROOT = __import__('pathlib').Path(__file__).resolve().parent.parent

def discover_themes():
    """Discover built-in themes from themes/*/manifest.json (single source of truth)."""
    themes_dir = ROOT / 'themes'
    if not themes_dir.is_dir():
        raise RuntimeError(f'themes directory missing: {themes_dir}')
    found = []
    for entry in sorted(themes_dir.iterdir()):
        if not entry.is_dir() or entry.name.startswith('_'):
            continue
        try:
            manifest = json.loads((entry / 'manifest.json').read_text(encoding='utf-8'))
        except (OSError, ValueError):
            continue
        if isinstance(manifest, dict) and manifest.get('id') == entry.name:
            found.append(entry.name)
    return tuple(found)

THEMES = discover_themes()
ALLOWED_TAGS = ['p','br','b','i','u','em','strong','h1','h2','h3','h4','h5','h6','ul','ol','li','blockquote','pre','code','a','img','figure','figcaption','table','thead','tbody','tr','th','td','hr','div','span']
ALLOWED_ATTRS = {'a':['href','title','target','rel'], 'img':['src','alt','title','loading','width','height'], 'td':['colspan','rowspan'], 'th':['colspan','rowspan']}

def uid():
    return uuid.uuid4().hex[:12]

def sanitize(html):
    return bleach.clean(html or '', tags=ALLOWED_TAGS, attributes=ALLOWED_ATTRS, protocols=['http','https','mailto'], strip=True)

def route_path(value, default='/'):
    s = (value or default).strip()
    if s.startswith(('http://','https://')):
        s = urlparse(s).path or '/'
    # Query strings cannot be represented as ordinary static file routes.
    s = s.split('?',1)[0].split('#',1)[0]
    if not s.startswith('/'):
        s = '/' + s
    if '\\' in s or '\x00' in s or any(part in ('.','..') for part in unquote(s).split('/')):
        raise ValueError('不允许路径穿越或反斜杠')
    if s.startswith(('/api/','/preview/','/__siteforge/')):
        raise ValueError('该路径已保留给系统')
    if len(s)>500:
        raise ValueError('URL 路径过长')
    return s

def slug_of(title):
    s = re.sub(r'[^\w\-]+','-', title.strip().lower(), flags=re.UNICODE).strip('-')
    return s[:70] or uid()

def clean_hex(s):
    if not re.fullmatch(r'#[0-9a-fA-F]{6}', s or ''):
        raise ValueError('品牌色需要 #RRGGBB 格式')
    return s

def ensure_public_url(url):
    parsed=urlparse(url)
    if parsed.scheme not in ('https','http') or not parsed.hostname or parsed.username or parsed.password:
        raise ValueError('仅支持公开的 HTTP/HTTPS URL')
    hostname=parsed.hostname.lower()
    if hostname in ('localhost','localhost.localdomain') or hostname.endswith(('.local','.internal')):
        raise ValueError('禁止访问内网域名')
    try:
        addrs=socket.getaddrinfo(hostname, parsed.port or (443 if parsed.scheme=='https' else 80), type=socket.SOCK_STREAM)
        for info in addrs:
            ip=ipaddress.ip_address(info[4][0])
            if not ip.is_global:
                raise ValueError('不允许访问内网、回环或保留 IP 地址')
    except socket.gaierror:
        raise ValueError('无法解析域名')
    return url

def save_entry(site_id, data, existing_id=None, source='manual', replace_route=False):
    if not db.one('SELECT id FROM sites WHERE id=?',(site_id,)):
        raise ValueError('网站不存在')
    kind=str(data.get('type') or 'page').lower()
    if kind not in TYPES:
        kind='other'
    title=str(data.get('title') or '').strip()
    if not title:
        raise ValueError('内容标题不能为空')
    route=route_path(data.get('route'), '/' if kind=='page' and title.lower() in ('home','首页') else f'/{kind}/{slug_of(title)}/')
    if route=='/' and kind!='page':
        raise ValueError('首页必须是普通页面类型')
    old=db.one('SELECT * FROM entries WHERE site_id=? AND route=?',(site_id,route))
    if old and old['id'] != existing_id:
        if replace_route:
            existing_id=old['id']
        else:
            raise ValueError(f'URL 冲突：{route}')
    eid=existing_id or uid()
    prior=db.one('SELECT * FROM entries WHERE site_id=? AND id=?',(site_id,eid)) if existing_id else None
    if existing_id and not prior:
        raise ValueError('内容不存在')
    meta=data.get('meta', {})
    if isinstance(meta,str):
        try: meta=json.loads(meta)
        except json.JSONDecodeError: meta={}
    if not isinstance(meta,dict): meta={}
    if 'blocks' in meta:
        from .blueprints import validate_blocks
        meta['blocks']=validate_blocks(meta['blocks'])
    vals=(kind,title,route,slug_of(title),str(data.get('summary') or ''),sanitize(str(data.get('body') or '')),
          str(data.get('cover') or ''),str(data.get('seo_title') or ''),str(data.get('seo_description') or ''),
          str(data.get('canonical') or ''),str(data.get('robots') or 'index,follow'),json.dumps(meta,ensure_ascii=False),
          str(data.get('status') or 'published'))
    if prior:
        db.execute('''UPDATE entries SET type=?,title=?,route=?,slug=?,summary=?,body=?,cover=?,seo_title=?,seo_description=?,canonical=?,robots=?,meta_json=?,status=?,updated_at=? WHERE site_id=? AND id=?''',(*vals,db.now(),site_id,eid))
    else:
        db.insert('''INSERT INTO entries (type,title,route,slug,summary,body,cover,seo_title,seo_description,canonical,robots,meta_json,status,created_at,updated_at,site_id,id,source)
            VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)''',(*vals,db.now(),db.now(),site_id,eid,source))
    return eid

def entry_dict(e):
    v=dict(e)
    try:v['meta']=json.loads(v.pop('meta_json'))
    except (ValueError,TypeError):v['meta']={}
    return v