feat(ip): protect persona 'formula' — secret fields only super_admin sees/edits
IP protection so casual copying yields inferior results: - SECRET_PERSONA_FIELDS (pains/objections/negotiation_levers/opener/tolerance + pain rootCause/resolutionConditions): only super_admin can view/edit them. - list_personas/get_persona/update_persona/get_group strip these for role=admin (and hide sales_kit + pain-fit report from admins too). - update_persona rejects admin attempts to set secret fields (403). - PersonaForm hides the 'การขาย' recipe section for non-super-admin (shows locked note); auth.isSuperAdmin getter added. Rebuilt dist. Added test_ip_protection.
This commit is contained in:
@@ -14,6 +14,29 @@ from .helpers import ApiError, current_user, require_auth, require_roles
|
||||
|
||||
groups_bp = Blueprint("groups", __name__)
|
||||
|
||||
# Fields that encode the "formula"/process of a persona. Only super_admin may see/edit
|
||||
# them; admins get the persona but NOT these — so a casual copy yields inferior results.
|
||||
SECRET_PERSONA_FIELDS = {
|
||||
"pains", "objections", "negotiation_levers", "opener",
|
||||
"rootCause", "resolutionConditions", "tolerance",
|
||||
}
|
||||
|
||||
|
||||
def strip_secret_fields(persona: dict) -> dict:
|
||||
"""Return a copy of a persona with secret/process fields removed."""
|
||||
out = dict(persona)
|
||||
for f in SECRET_PERSONA_FIELDS:
|
||||
out.pop(f, None)
|
||||
pains = out.get("pains")
|
||||
if isinstance(pains, list):
|
||||
cleaned = []
|
||||
for p in pains:
|
||||
if isinstance(p, dict):
|
||||
p = {k: v for k, v in p.items() if k not in ("rootCause", "resolutionConditions")}
|
||||
cleaned.append(p)
|
||||
out["pains"] = cleaned
|
||||
return out
|
||||
|
||||
_ANALYZE_LOCKS: dict[str, threading.Lock] = {}
|
||||
_ANALYZE_GUARD = threading.Lock()
|
||||
|
||||
@@ -203,6 +226,12 @@ def get_group(gid: str):
|
||||
view["personas"] = [revealable_view(p) for p in group.get("personas", [])]
|
||||
view["sales_kit"] = None
|
||||
view["report"] = None
|
||||
elif actor.get("role") != "super_admin":
|
||||
# admin: see group, but not secret/process persona fields nor the sales-kit/report
|
||||
# (pain-fit analysis is the IP to protect).
|
||||
view["personas"] = [strip_secret_fields(p) for p in group.get("personas", [])]
|
||||
view["sales_kit"] = None
|
||||
view["report"] = None
|
||||
return jsonify({"group": view})
|
||||
|
||||
|
||||
@@ -216,8 +245,11 @@ def list_personas(gid: str):
|
||||
if group.get("status") != "ready":
|
||||
raise ApiError("group not ready", 403)
|
||||
personas = [revealable_view(p) for p in group.get("personas", [])]
|
||||
else:
|
||||
elif actor.get("role") == "super_admin":
|
||||
personas = group.get("personas", [])
|
||||
else:
|
||||
# admin: see persona but not the secret/process fields (IP protection)
|
||||
personas = [strip_secret_fields(p) for p in group.get("personas", [])]
|
||||
# attach per-user status (won/lost/not-tried) for trainees
|
||||
if actor.get("role") == "user":
|
||||
sess = _stores().get("session_store")
|
||||
@@ -244,7 +276,10 @@ def get_persona(gid: str, pid: str):
|
||||
ensure = ensure_persona_shape(p)
|
||||
if actor.get("role") == "user":
|
||||
return jsonify({"persona": revealable_view(ensure)})
|
||||
return jsonify({"persona": ensure})
|
||||
if actor.get("role") == "super_admin":
|
||||
return jsonify({"persona": ensure})
|
||||
# admin: hidden secret/process fields (IP protection)
|
||||
return jsonify({"persona": strip_secret_fields(ensure)})
|
||||
|
||||
|
||||
@groups_bp.put("/<gid>/personas/<pid>")
|
||||
@@ -254,13 +289,22 @@ def update_persona(gid: str, pid: str):
|
||||
s = _stores()
|
||||
_get_owned_group(s, gid)
|
||||
data = request.get_json(silent=True) or {}
|
||||
actor = current_user()
|
||||
# IP protection: only super_admin may set/alter secret formula fields.
|
||||
if actor.get("role") != "super_admin":
|
||||
for f in SECRET_PERSONA_FIELDS:
|
||||
if f in data:
|
||||
raise ApiError(f"field '{f}' is locked (super_admin only)", 403)
|
||||
try:
|
||||
updated = s["groups"].update_persona(gid, pid, data)
|
||||
except ValueError as exc:
|
||||
raise ApiError(str(exc), 404)
|
||||
return jsonify({"persona": ensure_persona_shape(updated["personas"][
|
||||
full = ensure_persona_shape(updated["personas"][
|
||||
next(i for i, p in enumerate(updated["personas"]) if p["id"] == pid)
|
||||
])})
|
||||
])
|
||||
if actor.get("role") == "super_admin":
|
||||
return jsonify({"persona": full})
|
||||
return jsonify({"persona": strip_secret_fields(full)})
|
||||
|
||||
|
||||
@groups_bp.post("/<gid>/reanalyze")
|
||||
|
||||
61
backend/scripts/test_ip_protection.py
Normal file
61
backend/scripts/test_ip_protection.py
Normal file
@@ -0,0 +1,61 @@
|
||||
"""Test: IP protection — admin cannot see/edit secret persona fields; super_admin can."""
|
||||
import os, sys, tempfile, warnings
|
||||
from pathlib import Path
|
||||
|
||||
warnings.filterwarnings("ignore")
|
||||
BACKEND = str(Path(__file__).resolve().parents[1])
|
||||
sys.path.insert(0, BACKEND)
|
||||
|
||||
from app.factory import create_app
|
||||
from app.config import Config
|
||||
|
||||
td = tempfile.mkdtemp()
|
||||
Config.DATA_DIR = Path(td)
|
||||
sys.path.insert(0, BACKEND + "/scripts")
|
||||
from mock_llm import MockLLM
|
||||
|
||||
app = create_app()
|
||||
app.extensions["llm"] = MockLLM()
|
||||
C = app.test_client()
|
||||
|
||||
def tok(u, p): return C.post("/api/auth/login", json={"username": u, "password": p}).get_json()["token"]
|
||||
|
||||
AT = tok("admin", "1234"); AH = {"Authorization": f"Bearer {AT}"}
|
||||
C.post("/api/auth/setup", headers=AH, json={"username": "admin", "email": "a@b.co", "password": "newpass"})
|
||||
AT = tok("admin", "newpass"); AH = {"Authorization": f"Bearer {AT}"}
|
||||
|
||||
gid = C.post("/api/groups", headers=AH, json={"product": "CRM", "segment": "SME", "channel": "line", "language": "th"}).get_json()["group"]["id"]
|
||||
C.post(f"/api/groups/{gid}/analyze", headers=AH)
|
||||
personas = C.get(f"/api/groups/{gid}/personas", headers=AH).get_json()["personas"] # super_admin sees all
|
||||
assert personas and "pains" in personas[0], "super_admin should see secret fields"
|
||||
print("[ok] super_admin sees secret fields")
|
||||
|
||||
# create an admin (not super) user
|
||||
C.post("/api/admin/users", headers=AH, json={"username": "adm", "name": "Adm", "password": "pppp", "role": "admin"})
|
||||
AT2 = tok("adm", "pppp"); AH2 = {"Authorization": f"Bearer {AT2}"}
|
||||
pid = personas[0]["id"]
|
||||
|
||||
# admin list_personas: secret fields stripped
|
||||
admin_list = C.get(f"/api/groups/{gid}/personas", headers=AH2).get_json()["personas"]
|
||||
assert "pains" not in admin_list[0] and "tolerance" not in admin_list[0], "admin list should strip secrets"
|
||||
print("[ok] admin list strips secret fields")
|
||||
|
||||
# admin update with a secret field -> 403
|
||||
r = C.put(f"/api/groups/{gid}/personas/{pid}", headers=AH2, json={"name": "X", "tolerance": 1})
|
||||
assert r.status_code == 403, r.get_json()
|
||||
print("[ok] admin cannot set secret field (403)")
|
||||
|
||||
# admin update of non-secret field -> ok, but response still strips secrets
|
||||
r = C.put(f"/api/groups/{gid}/personas/{pid}", headers=AH2, json={"name": "Edited Name"})
|
||||
assert r.status_code == 200, r.get_json()
|
||||
body = r.get_json()["persona"]
|
||||
assert body["name"] == "Edited Name"
|
||||
assert "pains" not in body, "admin update response should strip secrets"
|
||||
print("[ok] admin can edit non-secret field; response strips secrets")
|
||||
|
||||
# super_admin can update secret field
|
||||
r = C.put(f"/api/groups/{gid}/personas/{pid}", headers=AH, json={"tolerance": 5})
|
||||
assert r.status_code == 200, r.get_json()
|
||||
print("[ok] super_admin can edit secret field")
|
||||
|
||||
print("ALL IP-PROTECTION TESTS PASSED")
|
||||
Reference in New Issue
Block a user