refs #699: align HMM portal audit experience

poc4_active_source_20260714/apps/poc4/mcp_discovery_ui.py

@@ -62,6 +62,7 @@ from src.agent_console.presentation import (
     render_console_header,
     render_login_brand,
 )
+from src.agent_console.audit import render_hmm_audit_tab
 from src.agent_console.profile import AppProfile, AppProfileError, load_app_profile
 from src.poc4.scenarios import ScenarioConfigError, load_demo_scenarios
 
@@ -93,10 +94,16 @@ PORTAL_AUTH_USER_KEY = "poc4_portal_auth_user"
 PORTAL_LOGIN_FAILURE_KEY = "poc4_portal_login_failed"
 PORTAL_REMEMBER_TOKEN_PARAM = "poc4_remember"
 PORTAL_REMEMBER_MAX_AGE_SECONDS = 7 * 24 * 60 * 60
-AUDIT_SCHEMA = "POC_2"
 AUDIT_DB_ENV_FILE = Path(
-    os.environ.get("POC4_AUDIT_DB_ENV_FILE", "/home/opc/kbmcp/.env")
+    os.environ.get("POC4_AUDIT_DB_ENV_FILE", str(ENV_FILE))
 ).expanduser()
+DEFAULT_AUDIT_DB_DSN = (
+    "(description=(retry_count=3)(retry_delay=1)"
+    "(address=(protocol=tcps)(port=1521)(host=adb.ap-seoul-1.oraclecloud.com))"
+    "(connect_data=(service_name="
+    "yh0olybn5pqce4n_hmmaipoc_high.adb.oraclecloud.com))"
+    "(security=(ssl_server_dn_match=yes)))"
+)
 _OPAQUE_BEARER = re.compile(r"^[\x21-\x7e]{1,4096}$")
 KB_THEME_CSS = """
 <style>
@@ -2147,29 +2154,40 @@ def _audit_db_env_value(name: str, default: str = "") -> str:
 
 @st.cache_resource(show_spinner=False)
 def _audit_db_pool() -> Any:
-    password = _audit_db_env_value("ORACLE_DB_PASSWORD")
+    password = _audit_db_env_value("POC4_AUDIT_DB_PASSWORD")
     if not password:
         raise AuditLogError("감사로그 DB 접속 설정을 확인해 주세요.")
-    wallet_dir = Path(
-        _audit_db_env_value(
-            "ORACLE_WALLET_DIR",
-            "/home/opc/wallet/kbaipoc",
-        )
-    ).expanduser().resolve()
-    if not wallet_dir.is_dir():
-        raise AuditLogError("감사로그 DB Wallet 경로를 확인해 주세요.")
-    try:
-        return oracledb.create_pool(
-            user=_audit_db_env_value("ORACLE_DB_USER", "ADMIN"),
-            password=password,
-            dsn=_audit_db_env_value("ORACLE_DSN", "kbaipoc_high"),
+    dsn = _audit_db_env_value("POC4_AUDIT_DSN", DEFAULT_AUDIT_DB_DSN)
+    wallet_password = _audit_db_env_value("POC4_AUDIT_WALLET_PASSWORD")
+    pool_options: dict[str, Any] = {}
+    if wallet_password:
+        wallet_dir = Path(
+            _audit_db_env_value(
+                "POC4_AUDIT_WALLET_DIR",
+                "/home/opc/apps/vpd-backoffice/wallet",
+            )
+        ).expanduser().resolve()
+        if not wallet_dir.is_dir():
+            raise AuditLogError("감사로그 DB Wallet 경로를 확인해 주세요.")
+        pool_options.update(
             config_dir=str(wallet_dir),
             wallet_location=str(wallet_dir),
-            wallet_password=_audit_db_env_value("ORACLE_WALLET_PASSWORD") or None,
+            wallet_password=wallet_password,
+        )
+    elif not (dsn.lstrip().startswith("(") or dsn.lower().startswith("tcps://")):
+        raise AuditLogError(
+            "감사로그 DB DSN은 TLS 접속 기술자이거나 Wallet 암호와 함께 제공되어야 합니다."
+        )
+    try:
+        return oracledb.create_pool(
+            user=_audit_db_env_value("POC4_AUDIT_DB_USER", "ADMIN"),
+            password=password,
+            dsn=dsn,
             min=1,
             max=2,
             increment=1,
             getmode=oracledb.POOL_GETMODE_WAIT,
+            **pool_options,
         )
     except (oracledb.Error, OSError, ValueError):
         raise AuditLogError("감사로그 DB에 연결하지 못했습니다.") from None
@@ -2200,111 +2218,59 @@ def _audit_rows(
 
 
 @st.cache_data(ttl=60, show_spinner=False)
-def _load_fga_inventory() -> dict[str, list[dict[str, Any]]]:
-    policies = _audit_rows(
+def _load_hmm_audit_inventory() -> list[dict[str, Any]]:
+    return _audit_rows(
         """
-        SELECT policy.object_name,
-               policy.policy_name,
-               policy.policy_text,
-               LISTAGG(policy_columns.policy_column, ',') WITHIN GROUP (
-                   ORDER BY policy_columns.policy_column
-               ) AS policy_column,
-               policy.enabled,
-               policy.sel,
-               policy.ins,
-               policy.upd,
-               policy.del
-        FROM dba_audit_policies policy
-        LEFT JOIN dba_audit_policy_columns policy_columns
-          ON policy_columns.object_schema = policy.object_schema
-         AND policy_columns.object_name = policy.object_name
-         AND policy_columns.policy_name = policy.policy_name
-        WHERE policy.object_schema = :schema
-        GROUP BY policy.object_name,
-                 policy.policy_name,
-                 policy.policy_text,
-                 policy.enabled,
-                 policy.sel,
-                 policy.ins,
-                 policy.upd,
-                 policy.del
-        ORDER BY policy.object_name, policy.policy_name
-        """,
-        {"schema": AUDIT_SCHEMA},
-    )
-    catalog = _audit_rows(
-        """
-        SELECT object_name,
-               column_name,
-               policy_name,
-               policy_expression,
-               enabled_yn,
-               description,
-               updated_at
-        FROM POC_2.KB_SECURITY_POLICY_CATALOG
-        WHERE control_type = 'DBMS_FGA'
-        ORDER BY object_name, policy_name, column_name
+        SELECT event_type,
+               COUNT(*) AS event_count,
+               TO_CHAR(
+                   MAX(created_at) AT TIME ZONE 'Asia/Seoul',
+                   'YYYY-MM-DD HH24:MI:SS'
+               ) AS latest_event_time
+        FROM ADMIN.HMM_ACCESS_AUDIT
+        GROUP BY event_type
+        ORDER BY event_type
         """
     )
-    return {"policies": policies, "catalog": catalog}
 
 
 @st.cache_data(ttl=30, show_spinner=False)
-def _load_fga_audit_events(
+def _load_hmm_audit_events(
     days: int,
     row_limit: int,
-    policy_name: str,
-    object_name: str,
+    event_type: str,
+    status: str,
 ) -> list[dict[str, Any]]:
     return _audit_rows(
         """
         SELECT *
         FROM (
-            SELECT TO_CHAR(
-                       audit_event.event_timestamp AT TIME ZONE 'Asia/Seoul',
+            SELECT audit_id,
+                   TO_CHAR(
+                       created_at AT TIME ZONE 'Asia/Seoul',
                        'YYYY-MM-DD HH24:MI:SS'
                    ) AS event_time,
-                   audit_event.dbusername,
-                   audit_event.client_identifier,
-                   audit_event.userhost,
-                   audit_event.object_schema,
-                   audit_event.object_name,
-                   audit_event.action_name,
-                   audit_event.fga_policy_name,
-                   policy_columns.audit_column,
-                   audit_event.return_code,
-                   DBMS_LOB.SUBSTR(audit_event.sql_text, 1000, 1) AS sql_text
-            FROM unified_audit_trail audit_event
-            LEFT JOIN (
-                SELECT object_schema,
-                       object_name,
-                       policy_name,
-                       LISTAGG(policy_column, ',') WITHIN GROUP (
-                           ORDER BY policy_column
-                       ) AS audit_column
-                FROM dba_audit_policy_columns
-                WHERE object_schema = :schema
-                GROUP BY object_schema, object_name, policy_name
-            ) policy_columns
-              ON policy_columns.object_schema = audit_event.object_schema
-             AND policy_columns.object_name = audit_event.object_name
-             AND policy_columns.policy_name = audit_event.fga_policy_name
-            WHERE audit_event.object_schema = :schema
-              AND audit_event.fga_policy_name IS NOT NULL
-              AND audit_event.event_timestamp >= (
-                  SYSTIMESTAMP - NUMTODSINTERVAL(:days, 'DAY')
-              )
-              AND (:policy_name IS NULL OR audit_event.fga_policy_name = :policy_name)
-              AND (:object_name IS NULL OR audit_event.object_name = :object_name)
-            ORDER BY audit_event.event_timestamp DESC
+                   event_type,
+                   key_id,
+                   object_id,
+                   status,
+                   row_count,
+                   error_code,
+                   message
+            FROM ADMIN.HMM_ACCESS_AUDIT
+            WHERE created_at >= (
+                SYSTIMESTAMP - NUMTODSINTERVAL(:days, 'DAY')
+            )
+              AND (:event_type IS NULL OR event_type = :event_type)
+              AND (:status IS NULL OR status = :status)
+            ORDER BY created_at DESC, audit_id DESC
         )
         WHERE ROWNUM <= :row_limit
         """,
         {
-            "schema": AUDIT_SCHEMA,
             "days": int(days),
-            "policy_name": policy_name or None,
-            "object_name": object_name or None,
+            "event_type": event_type or None,
+            "status": status or None,
             "row_limit": int(row_limit),
         },
     )
@@ -6116,280 +6082,6 @@ def _render_architecture_tab() -> None:
     )
 
 
-def _render_fga_audit_tab() -> None:
-    st.markdown(
-        '<div class="kb-section-title input" role="heading" aria-level="3">'
-        '감사로그 ( 오라클 <strong>FGA</strong> )'
-        '</div>',
-        unsafe_allow_html=True,
-    )
-    st.markdown(
-        '<div class="kb-audit-lead">'
-        'Oracle FGA 정책 상태와 민감 컬럼 접근 이력을 시간순으로 확인합니다. '
-        '조회 조건을 선택하면 정책·객체·사용자·SQL 원문을 함께 비교할 수 있습니다.'
-        '</div>',
-        unsafe_allow_html=True,
-    )
-    try:
-        inventory = _load_fga_inventory()
-    except AuditLogError as exc:
-        st.error(str(exc))
-        return
-
-    policies = inventory["policies"]
-    catalog = inventory["catalog"]
-    policy_names = sorted(
-        {
-            str(item.get("policy_name") or "").strip()
-            for item in (*policies, *catalog)
-            if str(item.get("policy_name") or "").strip()
-        }
-    )
-    object_names = sorted(
-        {
-            str(item.get("object_name") or "").strip()
-            for item in (*policies, *catalog)
-            if str(item.get("object_name") or "").strip()
-        }
-    )
-
-    st.markdown(
-        '<div class="kb-audit-heading">조회 조건</div>',
-        unsafe_allow_html=True,
-    )
-    with st.container(key="poc4_fga_filters"):
-        filter_policy, filter_object = st.columns(2)
-        with filter_policy:
-            selected_policy = st.selectbox(
-                "FGA 정책",
-                options=("", *policy_names),
-                format_func=lambda value: "전체 정책" if not value else value,
-                key="poc4_fga_policy_filter",
-            )
-        with filter_object:
-            selected_object = st.selectbox(
-                "감사 객체",
-                options=("", *object_names),
-                format_func=lambda value: "전체 객체" if not value else value,
-                key="poc4_fga_object_filter",
-            )
-        filter_days, filter_limit, refresh_column = st.columns([1.5, 1, 0.8])
-        with filter_days:
-            days = st.slider(
-                "조회 기간",
-                min_value=1,
-                max_value=90,
-                value=7,
-                format="%d일",
-                key="poc4_fga_days",
-            )
-        with filter_limit:
-            row_limit = st.number_input(
-                "최대 건수",
-                min_value=10,
-                max_value=500,
-                value=100,
-                step=10,
-                key="poc4_fga_row_limit",
-            )
-        with refresh_column:
-            st.markdown(
-                '<div style="height: 28px"></div>',
-                unsafe_allow_html=True,
-            )
-            if st.button(
-                "새로고침",
-                icon=":material/refresh:",
-                width="stretch",
-                key="poc4_fga_refresh",
-            ):
-                _load_fga_inventory.clear()
-                _load_fga_audit_events.clear()
-                st.rerun()
-
-    try:
-        events = _load_fga_audit_events(
-            int(days),
-            int(row_limit),
-            selected_policy,
-            selected_object,
-        )
-    except AuditLogError as exc:
-        st.error(str(exc))
-        return
-
-    success_count = sum(
-        1 for item in events if int(item.get("return_code") or 0) == 0
-    )
-    failure_count = len(events) - success_count
-    with st.container(key="poc4_fga_metrics"):
-        metric_policy, metric_event, metric_success, metric_failure = st.columns(4)
-        metric_policy.metric("등록 FGA 정책", len(policies))
-        metric_event.metric("조회 이벤트", len(events))
-        metric_success.metric("성공", success_count)
-        metric_failure.metric("실패", failure_count)
-
-    if not policies:
-        if catalog:
-            st.warning(
-                "FGA 정책 카탈로그는 존재하지만 현재 DB에 활성 정책이 등록되어 있지 않습니다."
-            )
-        else:
-            st.warning(
-                f"현재 {AUDIT_SCHEMA} 스키마에 등록된 DBMS_FGA 정책이 없습니다."
-            )
-
-    st.markdown(
-        '<div class="kb-audit-heading">정책 상태</div>'
-        f'<div class="kb-audit-caption">현재 활성 정책 {len(policies)}건 · '
-        '감사 대상 컬럼과 적용 조건을 확인합니다.</div>',
-        unsafe_allow_html=True,
-    )
-    with st.container(key="poc4_fga_policy_panel"):
-        with st.expander("FGA 정책 상태", expanded=True):
-            if policies:
-                st.dataframe(
-                    [
-                        {
-                            "객체": str(item.get("object_name") or ""),
-                            "정책": str(item.get("policy_name") or ""),
-                            "감사 컬럼": str(
-                                item.get("policy_column") or "전체"
-                            ),
-                            "조건": str(item.get("policy_text") or "항상"),
-                            "활성": str(item.get("enabled") or ""),
-                            "SELECT": str(item.get("sel") or ""),
-                        }
-                        for item in policies
-                    ],
-                    column_config={
-                        "객체": st.column_config.TextColumn(width="medium"),
-                        "정책": st.column_config.TextColumn(width="large"),
-                        "감사 컬럼": st.column_config.TextColumn(width="large"),
-                        "조건": st.column_config.TextColumn(width="large"),
-                        "활성": st.column_config.TextColumn(width="small"),
-                        "SELECT": st.column_config.TextColumn(width="small"),
-                    },
-                    hide_index=True,
-                    width="stretch",
-                    height=min(360, 72 + 36 * len(policies)),
-                )
-            elif catalog:
-                st.dataframe(
-                    [
-                        {
-                            "객체": str(item.get("object_name") or ""),
-                            "정책": str(item.get("policy_name") or ""),
-                            "대상 컬럼": str(item.get("column_name") or "전체"),
-                            "카탈로그 상태": str(item.get("enabled_yn") or ""),
-                            "설명": str(item.get("description") or ""),
-                        }
-                        for item in catalog
-                    ],
-                    hide_index=True,
-                    width="stretch",
-                    height=min(360, 72 + 36 * len(catalog)),
-                )
-            else:
-                st.caption("등록된 FGA 정책 정보가 없습니다.")
-
-    st.markdown(
-        '<div class="kb-audit-heading">감사 이벤트</div>'
-        '<div class="kb-audit-caption">최신 이벤트부터 표시합니다. '
-        '성공 여부와 감사 컬럼, 실행 사용자를 먼저 확인하세요.</div>',
-        unsafe_allow_html=True,
-    )
-    with st.container(key="poc4_fga_event_toolbar"):
-        show_sql = st.toggle(
-            "SQL 원문 표시",
-            value=True,
-            key="poc4_fga_show_sql",
-        )
-    if not events:
-        st.info("선택한 조건에 해당하는 FGA 감사 이벤트가 없습니다.")
-        return
-
-    display_rows: list[dict[str, Any]] = []
-    for event in events:
-        return_code = int(event.get("return_code") or 0)
-        actor = str(event.get("client_identifier") or "").strip()
-        if not actor:
-            actor = str(event.get("dbusername") or "")
-        display_row: dict[str, Any] = {
-            "발생시각(KST)": str(event.get("event_time") or ""),
-            "정책": str(event.get("fga_policy_name") or ""),
-            "감사 컬럼": str(event.get("audit_column") or "전체"),
-            "사용자": actor,
-            "DB 사용자": str(event.get("dbusername") or ""),
-            "접속 호스트": str(event.get("userhost") or ""),
-            "객체": (
-                f"{event.get('object_schema')}.{event.get('object_name')}"
-            ),
-            "작업": str(event.get("action_name") or ""),
-            "결과": "성공" if return_code == 0 else f"ORA-{return_code:05d}",
-        }
-        if show_sql:
-            display_row["SQL 원문"] = " ".join(
-                str(event.get("sql_text") or "").split()
-            )
-        display_rows.append(display_row)
-
-    def initial_column_width(
-        column_name: str,
-        minimum: int,
-        maximum: int,
-    ) -> int:
-        values = [column_name]
-        values.extend(str(row.get(column_name) or "") for row in display_rows)
-        text_units = max(
-            sum(2 if ord(character) > 127 else 1 for character in value)
-            for value in values
-        )
-        return max(minimum, min(maximum, 36 + text_units * 8))
-
-    event_column_config: dict[str, Any] = {
-        "발생시각(KST)": st.column_config.TextColumn(
-            width=initial_column_width("발생시각(KST)", 180, 220)
-        ),
-        "정책": st.column_config.TextColumn(
-            width=initial_column_width("정책", 180, 320)
-        ),
-        "감사 컬럼": st.column_config.TextColumn(
-            width=initial_column_width("감사 컬럼", 150, 300)
-        ),
-        "사용자": st.column_config.TextColumn(
-            width=initial_column_width("사용자", 110, 180)
-        ),
-        "DB 사용자": st.column_config.TextColumn(
-            width=initial_column_width("DB 사용자", 120, 180)
-        ),
-        "접속 호스트": st.column_config.TextColumn(
-            width=initial_column_width("접속 호스트", 150, 240)
-        ),
-        "객체": st.column_config.TextColumn(
-            width=initial_column_width("객체", 180, 300)
-        ),
-        "작업": st.column_config.TextColumn(
-            width=initial_column_width("작업", 90, 140)
-        ),
-        "결과": st.column_config.TextColumn(
-            width=initial_column_width("결과", 90, 140)
-        ),
-    }
-    if show_sql:
-        event_column_config["SQL 원문"] = st.column_config.TextColumn(
-            width=initial_column_width("SQL 원문", 420, 720)
-        )
-    with st.container(key="poc4_fga_event_panel"):
-        st.dataframe(
-            display_rows,
-            column_config=event_column_config,
-            hide_index=True,
-            width="stretch",
-            height=min(640, 104 + 38 * len(display_rows)),
-        )
-
-
 def _render_vpd_operations_tab() -> None:
     st.markdown(
         f"""
@@ -7487,7 +7179,12 @@ def main() -> None:
             )
 
     with audit_tab:
-        _render_fga_audit_tab()
+        render_hmm_audit_tab(
+            st,
+            _load_hmm_audit_inventory,
+            _load_hmm_audit_events,
+            AuditLogError,
+        )
 
     with operations_tab:
         _render_vpd_operations_tab()