Merge remote-tracking branch 'origin/dev' into sub_laptop_1

M02_MasterTemplete/M02_Layer_Access.py

@@ -1,13 +1,12 @@
 """M02 층 권한 한 벌 — 나무 · 항목(`Router_Items`) · 양식 · 남의 층 목록(`Router_Layers`)이 함께 씀
 (PLAN 27-1c). 화면 쪽 짝 = `M02_MasterTemplete_Layer.ts` 의 `accessFor`.
 
+「회사 C 를 맡음」(`manages` · 39-6) = SYSTEM_ADMIN · 또는 C 의 회사 관리자(ADMIN · 마스터).
     system        읽기 = 로그인 누구나 · 수정 = SYSTEM_ADMIN
-    company {C}   자기 회사 = 읽기 누구나 · 수정 = 회사 관리자(ADMIN · 마스터 · SYSTEM_ADMIN) ·
-                  남의 회사 = SYSTEM_ADMIN 읽기만
-    personal {C}/{U}  본인 = 읽기 · 수정 · 같은 회사 남 = 회사 관리자 읽기만 ·
-                  다른 회사 = SYSTEM_ADMIN 읽기만
-owner 글 없음 = 자기 층 · 남의 층은 그 층 폴더가 있을 때만(없으면 404) ·
-읽기로 남의 층에 뼈대를 안 만듦.
+    company {C}   읽기 = C 사람 · SYSTEM_ADMIN · 수정 = C 를 맡은 사람
+    personal {C}/{U}  본인 = 읽기 · 수정 · 남 = C 를 맡은 사람만 읽기 · 수정(일반 = 403)
+owner 글 없음 = 자기 층 · 남의 층 읽기는 그 층 폴더가 있을 때만(없으면 404) ·
+읽기로 남의 층에 뼈대를 안 만듦(쓰기는 부르는 쪽이 소속을 본 뒤 `folder(…, create=True)`).
 """
 
 from __future__ import annotations
@@ -33,6 +32,14 @@ def is_company_admin(session: dict[str, Any]) -> bool:
     return session.get("role") in ("ADMIN", "SYSTEM_ADMIN") or bool(session.get("is_master"))
 
 
+def manages(session: dict[str, Any], company: Any) -> bool:
+    """회사 `company` 를 맡음 — 그 회사 층 · 그 회사 모든 개인 층을 보고 고침."""
+    if is_system_admin(session):
+        return True
+    own = session.get("company_id")
+    return own is not None and str(own) == str(company) and is_company_admin(session)
+
+
 @dataclass(frozen=True)
 class Grant:
     """층 한 칸의 권한 — `owner` = 층 주인 글(시스템 None) · `mine` = 자기 층(뼈대 만들어도 됨)."""
@@ -63,15 +70,14 @@ def grant(session: dict[str, Any], layer: str, owner: str | None = None) -> Gran
         raise HTTPException(status_code=409, detail=NO_COMPANY)
     if not _SHAPE[layer].match(who):
         raise HTTPException(status_code=400, detail=f"층 주인 모양이 틀림 「{who}」")
+    company = who.split("/")[0]
     if who == own:
-        return Grant(layer, who, layer == "personal" or is_company_admin(session), True)
-    same_company = who.split("/")[0] == str(session.get("company_id"))
-    colleague = layer == "personal" and same_company and is_company_admin(session)
-    if not (is_system_admin(session) or colleague):
+        return Grant(layer, who, layer == "personal" or manages(session, company), True)
+    if not manages(session, company):
         raise HTTPException(
             status_code=403, detail="다른 회사 · 다른 사람의 구조물 집계표는 볼 수 없음"
         )
-    return Grant(layer, who, False, False)
+    return Grant(layer, who, True, False)
 
 
 def require(
@@ -82,16 +88,15 @@ def require(
     if write and not got.write:
         if layer == "system":
             detail = "마스터 템플릿은 시스템 관리자만 고침."
-        elif got.mine:
-            detail = "회사용은 회사 관리자만 고침 — 개인용으로 복사해서 씀."
         else:
-            detail = "남의 층은 읽기만."
+            detail = "회사용은 회사 관리자만 고침 — 개인용으로 복사해서 씀."
         raise HTTPException(status_code=403, detail=detail)
     return got
 
 
-def folder(got: Grant) -> Path:
-    """회사 · 개인 층 폴더 — 남의 층은 그 층 폴더가 있어야 함(부르는 쪽은 자기 층만 뼈대를 만듦)."""
+def folder(got: Grant, create: bool = False) -> Path:
+    """회사 · 개인 층 폴더 — 남의 층은 그 층 폴더가 있어야 함 · `create` = 쓰기(부르는 쪽이 소속을
+    본 뒤 뼈대를 만듦)."""
     if got.owner is None:
         raise HTTPException(status_code=409, detail=NO_COMPANY)
     try:
@@ -102,6 +107,6 @@ def folder(got: Grant) -> Path:
             path = layers.personal_dir(company, user)
     except ValueError as error:
         raise HTTPException(status_code=400, detail=f"층 주인 모양 — {error}") from error
-    if not got.mine and not path.is_dir():
+    if not (got.mine or create) and not path.is_dir():
         raise HTTPException(status_code=404, detail="그 층은 아직 만든 것이 없음")
     return path

M02_MasterTemplete/M02_MasterTemplete_Router_Combos.py

@@ -7,13 +7,14 @@
 저장소 = `M02_MasterTemplete_Combo_Store`.
 담을 로직 고르기 · 미리 보기가 읽는 단가산출 로직(M01 길은 시스템 관리자 전용)은
 `/api/m02/logic-lib/…` 읽기만(로그인) — 모양은 M01 길과 같음.
+39-6 — 일위대가 조합 = 실무문서 · 조합 길 전부 시스템 관리자만(`_door` · 담을 로직 읽기 뺌).
 """
 
 from __future__ import annotations
 
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel
 
 from common_util.common_util_auth import verify_session
@@ -24,7 +25,16 @@ from M02_MasterTemplete import M02_MasterTemplete_Combo_Copy as combo_copy
 from M02_MasterTemplete import M02_MasterTemplete_Combo_Store as combos
 from M02_MasterTemplete.M02_MasterTemplete_Router_Items import Layer, Session, _call, _root
 
-router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Combos"])
+
+def _door(request: Request, session: Session = Depends(verify_session)) -> None:
+    """일위대가 조합 = 실무문서(39-6) — 조합 길 전부 시스템 관리자만 · 담을 로직 읽기는 로그인만."""
+    if "/logic-lib/" not in request.url.path and not access.is_system_admin(session):
+        raise HTTPException(status_code=403, detail="실무문서(일위대가 조합)는 시스템 관리자만 봄.")
+
+
+router = APIRouter(
+    prefix="/api/m02", tags=["M02 MasterTemplete Combos"], dependencies=[Depends(_door)]
+)
 
 
 class ComboBody(BaseModel):

M02_MasterTemplete/M02_MasterTemplete_Router_Items.py

@@ -10,13 +10,17 @@ PLAN 27-1c) · `owner` = 회사 `{C}` · 개인 `{C}/{U}`(없으면 자기 층)
 
 from __future__ import annotations
 
+import logging
 from pathlib import Path
 from typing import Any, Literal
 
+import anyio.from_thread
 from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import BaseModel
 
+from common_util.common_util_audit import record_audit
 from common_util.common_util_auth import verify_session
+from config.config_db import get_db_pool
 from M02_MasterTemplete import M02_Item_Paste as paste_store
 from M02_MasterTemplete import M02_Layer_Access as access
 from M02_MasterTemplete import M02_MasterTemplete_Branch as branch
@@ -32,6 +36,7 @@ from M02_MasterTemplete.M02_MasterTemplete_Router import (
     UseComboBody,
 )
 
+logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Items"])
 
 Layer = Literal["system", "company", "personal"]
@@ -74,14 +79,60 @@ class CopyItemBody(BaseModel):
     key: str
 
 
+async def _member(owner: str) -> bool:
+    """층 주인이 있나 — 회사 `{C}` = 그 회사 · 개인 `{C}/{U}` = U 가 C 소속."""
+    company, _, user = owner.partition("/")
+    pool = get_db_pool()
+    async with pool.acquire() as connection, connection.cursor() as cursor:
+        if user:
+            await cursor.execute(
+                "SELECT 1 FROM users WHERE id = %s AND company_id = %s AND deleted_at IS NULL",
+                (int(user), int(company)),
+            )
+        else:
+            await cursor.execute(
+                "SELECT 1 FROM companies WHERE id = %s AND deleted_at IS NULL", (int(company),)
+            )
+        return await cursor.fetchone() is not None
+
+
+async def _note(session: Session, got: access.Grant) -> None:
+    """남의 층 쓰기 한 줄(39-6 · 시스템 로그) — 기록이 못 돼도 쓰기는 막지 않음."""
+    try:
+        pool = get_db_pool()
+        async with pool.acquire() as connection, connection.cursor() as cursor:
+            await record_audit(
+                cursor,
+                actor_id=int(session["user_id"]),
+                action="M02_OTHER_LAYER_WRITE",
+                resource_type=f"m02_{got.layer}",
+                resource_ref=got.owner,
+            )
+            await connection.commit()
+    except Exception:
+        logger.exception("M02 남의 층 쓰기 기록 실패: %s %s", got.layer, got.owner)
+
+
+async def others_write(session: Session, got: access.Grant, folder: Path) -> None:
+    """남의 층 쓰기 — 처음(집계표 없음)이면 주인이 있는지 본 뒤 뼈대 · 쓸 때마다 시스템 로그."""
+    if not (folder / "table" / f"{store.STRUCTURE_TABLE}.json").is_file():
+        if not await _member(str(got.owner)):
+            raise HTTPException(status_code=403, detail="그 회사 · 그 회사 사람이 아님")
+        items_store.ensure(folder)
+    await _note(session, got)
+
+
 def _root(session: Session, layer: str, owner: str | None = None, write: bool = False) -> Path:
-    """층 폴더(권한까지 · `M02_Layer_Access`) — 자기 회사 · 개인 층은 없으면 빈 뼈대를 만듦."""
+    """층 폴더(권한까지 · `M02_Layer_Access`) — 자기 회사 · 개인 층은 없으면 빈 뼈대를 만듦 ·
+    남의 층 쓰기(맡은 회사 · 39-6)는 `others_write`."""
     got = access.require(session, layer, owner, write=write)
     if layer == "system":
         return store.FOLDER
-    folder = access.folder(got)
+    folder = access.folder(got, create=write)
     if got.mine:
         items_store.ensure(folder)
+    elif write:
+        anyio.from_thread.run(others_write, session, got, folder)
     elif not (folder / "table" / f"{store.STRUCTURE_TABLE}.json").is_file():
         raise HTTPException(status_code=404, detail="그 층은 아직 만든 것이 없음")
     return folder

M02_MasterTemplete/M02_MasterTemplete_Router_Layers.py

@@ -27,6 +27,7 @@ from common_util.common_util_storage import resolve_stored_project_path
 from config.config_db import get_db_pool, run_with_connection
 from M02_MasterTemplete import M02_Layer_Access as access
 from M02_MasterTemplete import M02_MasterTemplete_Items as items_store
+from M02_MasterTemplete import M02_MasterTemplete_Router_Items as items_router
 from M02_MasterTemplete import M02_MasterTemplete_Store as store
 from M02_MasterTemplete import M02_Table_Fill as fill
 from M02_MasterTemplete import M02_Template_Layers as layers
@@ -164,8 +165,10 @@ def _company_of(session: dict[str, Any], project: dict[str, Any] | None) -> int:
 
 
 async def _same_company_user(session: dict[str, Any], user_id: int, company_id: int) -> None:
+    """남의 개인 양식 = 그 회사를 맡은 사람만(39-6 · 층 길과 한 규칙) · 그 회사 사람이어야 함."""
     if user_id == session.get("user_id"):
         return
+    access.grant(session, "personal", f"{company_id}/{user_id}")
     if await _user_company(user_id) != company_id:
         raise HTTPException(status_code=403, detail="같은 회사 사람의 양식만 볼 수 있습니다.")
 
@@ -192,7 +195,12 @@ async def _layer_dir(
         who = session["user_id"] if user_id is None else user_id
         owner = str(company_id) if layer == "company" else f"{company_id}/{who}"
     got = access.require(session, layer, owner, write=write)
-    return _ensured(access.folder(got)) if got.mine else access.folder(got)
+    folder = access.folder(got, create=write)
+    if got.mine:
+        return _ensured(folder)
+    if write:  # 남의 층 쓰기(맡은 회사 · 39-6) — 구조물 길과 한 벌
+        await items_router.others_write(session, got, folder)
+    return folder
 
 
 def _ensured(folder: Path) -> Path:
@@ -214,9 +222,9 @@ async def list_owners(
     company: int | None = Query(None),
     session: dict[str, Any] = Depends(verify_session),
 ) -> list[dict[str, Any]]:
-    """남의 층 목록(PLAN 27-1c · 펼칠 때 부름) — `[{owner, name}]` · 층 자리가 있는 것만(본인 뺌).
-    회사 = SYSTEM_ADMIN 만 회사 목록 · 개인 = 그 회사 사람(SYSTEM_ADMIN 은 어느 회사나 ·
-    회사 관리자는 자기 회사만)."""
+    """남의 층 목록(PLAN 27-1c · 펼칠 때 부름) — `[{owner, name}]`.
+    회사 = SYSTEM_ADMIN 만 · 층 자리가 있는 것만(자기 회사는 늘) ·
+    개인 = 그 회사를 맡은 사람(39-6)만 · 회사 사람 전부(나 맨 위 · 집계표 없으면 `비어있음`)."""
     if layer == "company":
         if not access.is_system_admin(session):
             raise HTTPException(status_code=403, detail="회사 목록은 시스템 관리자만 봄.")
@@ -225,12 +233,16 @@ async def list_owners(
         target = session.get("company_id") if company is None else company
         if target is None:
             raise HTTPException(status_code=409, detail=access.NO_COMPANY)
-        mine = str(target) == str(session.get("company_id"))
-        if not (access.is_system_admin(session) or (mine and access.is_company_admin(session))):
+        if not access.manages(session, target):
             raise HTTPException(status_code=403, detail="같은 회사 사람 목록은 회사 관리자만 봄.")
         me = session.get("user_id")
         users = await _company_users(int(target))
-        rows = [(f"{target}/{u['user_id']}", u["name"]) for u in users if u["user_id"] != me]
+        users.sort(key=lambda u: u["user_id"] != me)  # 나 맨 위
+        return [
+            {"owner": owner, "name": u["name"], "비어있음": _empty(owner)}
+            for u in users
+            for owner in [f"{target}/{u['user_id']}"]
+        ]
     found = []
     for owner, name in rows:
         got = access.Grant(layer, owner, False, owner == access.own_owner(session, layer))
@@ -242,6 +254,13 @@ async def list_owners(
     return found
 
 
+def _empty(owner: str) -> bool:
+    """개인 층에 집계표가 없음 — 사람 나무 「비어 있음」(폴더를 안 만듦)."""
+    company, user = owner.split("/")
+    table = layers.personal_dir(company, user) / "table" / f"{store.STRUCTURE_TABLE}.json"
+    return not table.is_file()
+
+
 @router.get("/layers/{layer}/templates")
 async def list_layer(
     layer: Layer,
@@ -519,7 +538,11 @@ async def list_sources(
                 )
         return rows
 
-    users = await _company_users(company_id)
+    users = [  # 남의 개인 = 그 회사를 맡은 사람만(39-6)
+        u
+        for u in await _company_users(company_id)
+        if u["user_id"] == session.get("user_id") or access.manages(session, company_id)
+    ]
     projects = await _company_projects(company_id)
     return {
         "system": await asyncio.to_thread(layers.list_templates, layers.system_dir()),