Server to Server 암호화 가이드¶

INISAFE_NET/
├── docs/
│   ├── Server To Server_암호화가이드.pdf
│   ├── INISAFE_NET_C_to_Python_v7.2_Developer_Guide_v1.0.pdf
│   └── INISAFE_NET_for_C_v7.2_Developer_Guide_v1.13.pdf
└── INISAFE_Net_for_C_v7.2.47_64_Linux_4.18.0.tar.gz # INISAFE Net SDK

INISAFE_Net_for_C_v7.2.47_64_Linux_4.18.0/
    ├── conf/
    │   ├── INISAFENet_A.cnf      # INISAFE 설정 파일 A
    │   └── INISAFENet_B.cnf      # INISAFE 설정 파일 B
    ├── keys/
    │   ├── EncryptKey_A.der      # 세션 키 파일 A
    │   └── EncryptKey_B.der      # 세션 키 파일 B
    ├── lib/
    │   ├── libINISAFE_Crypto_for_C_v5.2.0_Linux_4.18_64.so
    │   ├── libINISAFE_PKI_for_C_v5.2.6_Linux_4.18_64.so
    │   ├── libiniCore_v2.4.6_Linux_4.18.so
    │   └── libinisafeNet_7.2.47_Linux-4.18.0-193.28.1.el8_2.x86_64.so
    ├── sample/
    │   └── keyfix
    │       ├── inl_constants.py
    │       ├── inl_functions.py
    │       ├── keyfix.py         # 테스트 코드
    │       ├── keyfix_cnfkey.py  # 테스트 코드
    │       └── keyfix_dualkey.py # 다중키 테스트 코드
    ├── license/
    │   └── default.lic         # INISAFE 라이선스
    ├── log/
    └── tools/
        ├── gen_ivkey           # 세션 키 생성 도구
        ├── encrypt_pwd         # 비밀번호 암호화 도구
        └── decrypt_pwd         # 비밀번호 복호화 도구

# 1. INISAFE Net 패키지 디렉토리로 이동
cd /path/to/INISAFE_Net_for_C_v7.2.47_64_Linux_4.18.0

# 2. Ubuntu 22.04 컨테이너 실행
#    현재 디렉토리를 컨테이너 내 /app으로 마운트
docker run -it -v "$(pwd):/app" --platform linux/amd64 ubuntu:22.04

# 3. Python 설치 (컨테이너 내부)
apt update && apt install -y python3

# 4. 스크립트 실행
INISAFENET_HOME=/app LD_LIBRARY_PATH=/app/lib python3 /app/sample/keyfix/keyfix_cnfkey.py

export INISAFENET_HOME=/path/to/inisafe
export LD_LIBRARY_PATH=$INISAFENET_HOME/lib:$LD_LIBRARY_PATH

volumes:
  - /path/to/inisafe:/app/inisafe
environment:
  INISAFENET_HOME: /app/inisafe
  LD_LIBRARY_PATH: /app/inisafe/lib

cd inisafe/lib/

ln -sf libINISAFE_Crypto_for_C_v5.2.0_Linux_4.18_64.so    libiniCrypto.so
ln -sf libiniCore_v2.4.6_Linux_4.18.so                     libiniCore.so
ln -sf libINISAFE_PKI_for_C_v5.2.6_Linux_4.18_64.so       libiniPKI.so
ln -sf libinisafeNet_7.2.47_Linux-4.18.0-193.28.1.el8_2.x86_64.so  libinisafeNet.so

cd $INISAFENET_HOME/tools
./gen_ivkey

1: old format(use Hash)
2: new format(use PBKDF2)
3: new format(use PBKDF2,sha256)
Enter the create version: 1

< iv > Input the 16 chracter of the number,
Enter New iv : 1234567890123456      ← 16자리 IV 입력

< skey > Input the 16 chracter of the number,
Enter New skey : INISAFE NETWORK.    ← 16자리 세션 키 입력

Enter New Password : ********        ← 세션키 비밀번호 입력

Success to create session-key file: EncryptKey.der

cd $INISAFENET_HOME/tools
./encrypt_pwd

Enter Password : ********        ← 세션키 비밀번호 입력

Inputed Password : [********]
Encrypted Password : [QZcawOC1VDkgRljCFNYeU2EBysLOWtdWk52r+mHgRu8=]

import ctypes
import threading
import base64
import os
import atexit
import weakref

# 의존성 순서대로 로딩
ctypes.CDLL("libiniCrypto.so", mode=ctypes.RTLD_GLOBAL)
ctypes.CDLL("libiniCore.so", mode=ctypes.RTLD_GLOBAL)
ctypes.CDLL("libiniPKI.so", mode=ctypes.RTLD_GLOBAL)
lib = ctypes.CDLL("libinisafeNet.so")

lib.INL_Initialize.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_char_p]
lib.INL_Initialize.restype = ctypes.c_int
lib.INL_New_Ctx.argtypes = [ctypes.c_int, ctypes.POINTER(ctypes.c_void_p)]
lib.INL_New_Ctx.restype = ctypes.c_int
lib.INL_Free_Ctx.argtypes = [ctypes.c_void_p]
lib.INL_Free_Ctx.restype = ctypes.c_int
lib.INL_Encrypt.argtypes = [
    ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int,
    ctypes.POINTER(ctypes.c_char_p), ctypes.POINTER(ctypes.c_int),
]
lib.INL_Encrypt.restype = ctypes.c_int
lib.INL_Decrypt.argtypes = [
    ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int,
    ctypes.POINTER(ctypes.c_char_p), ctypes.POINTER(ctypes.c_int),
]
lib.INL_Decrypt.restype = ctypes.c_int
lib.INL_Free_Buf.argtypes = [ctypes.c_char_p]
lib.INL_Free_Buf.restype = None
lib.INL_Clean.argtypes = []
lib.INL_Clean.restype = None
lib.INL_Error_String.argtypes = [ctypes.c_int]
lib.INL_Error_String.restype = ctypes.c_char_p

FIXKEY_CTX = 0x08
conf_path = os.path.join(os.environ["INISAFENET_HOME"], "conf", "INISAFENet.cnf")

ret = lib.INL_Initialize(FIXKEY_CTX, conf_path.encode("utf-8"), None)
if ret != 0:
    err = lib.INL_Error_String(ret)
    err_msg = err.decode("utf-8", errors="replace") if err else "unknown"
    raise RuntimeError(f"INL_Initialize failed: code={ret}, {err_msg}")

# 프로세스 종료 시 자동으로 리소스 정리
atexit.register(lib.INL_Clean)

_local = threading.local()

def _free_ctx(ctx: ctypes.c_void_p) -> None:
    if ctx.value is not None:
        lib.INL_Free_Ctx(ctx)
        ctx.value = None

def _get_ctx() -> ctypes.c_void_p:
    if not hasattr(_local, "ctx"):
        ctx = ctypes.c_void_p(None)
        ret = lib.INL_New_Ctx(FIXKEY_CTX, ctypes.byref(ctx))
        if ret != 0:
            raise RuntimeError(f"INL_New_Ctx failed: {ret}")
        _local.ctx = ctx

        # 메인 스레드는 atexit(INL_Clean)이 처리하므로 제외
        if threading.current_thread() is not threading.main_thread():
            weakref.finalize(
                threading.current_thread(),
                _free_ctx,
                ctx,
            )
    return _local.ctx

def encrypt(plaintext: bytes) -> bytes:

    ctx  = _get_ctx()
    out  = ctypes.c_char_p(None)
    outl = ctypes.c_int(0)

    try:
        ret = lib.INL_Encrypt(ctx, plaintext, len(plaintext),
                              ctypes.byref(out), ctypes.byref(outl))
        if ret != 0:
            raise RuntimeError(f"INL_Encrypt failed: {ret}")

        return out.value[:outl.value]
    finally:
        if out:
            lib.INL_Free_Buf(out)

def decrypt(ciphertext: bytes) -> bytes:

    ctx  = _get_ctx()
    out  = ctypes.c_char_p(None)
    outl = ctypes.c_int(0)

    try:
        ret = lib.INL_Decrypt(ctx, ciphertext, len(ciphertext),
                              ctypes.byref(out), ctypes.byref(outl))
        if ret != 0:
            raise RuntimeError(f"INL_Decrypt failed: {ret}")

        return out.value[:outl.value]
    finally:
        if out:
            lib.INL_Free_Buf(out)

# 평문
data: hi~

data: how

data: r u?

# 암호문 (실제 base64 값은 KeyFix 결과에 따라 달라진다)
data: AQpofiE=

data: AQhow==

data: AXIgdT8=

event: message
data: {base64(cipher)}

# 평문
{"content":"hi"}\n
{"content":"there"}\n

# 암호문
{base64(cipher)}\n
{base64(cipher)}\n

INISAFENET_HOME = os.environ["INISAFENET_HOME"]
conf_a = os.path.join(INISAFENET_HOME, "conf", "INISAFENet_A.cnf")  # Key A
conf_b = os.path.join(INISAFENET_HOME, "conf", "INISAFENet_B.cnf")  # Key B

# INL_Initialize 호출은 ctx 초기화에서 수행한다.

_init_lock         = threading.Lock()
_local             = threading.local()
_atexit_registered = False

def _free_ctx(ctx: ctypes.c_void_p) -> None:
    if ctx.value is not None:
        lib.INL_Free_Ctx(ctx)
        ctx.value = None

def _new_ctx(conf_path: str) -> ctypes.c_void_p:
    global _atexit_registered
    with _init_lock:
        ret = lib.INL_Initialize(FIXKEY_CTX, conf_path.encode("utf-8"), None)
        if ret != 0:
            err = lib.INL_Error_String(ret)
            err_msg = err.decode("utf-8", errors="replace") if err else "unknown"
            raise RuntimeError(f"INL_Initialize failed: code={ret}, {err_msg}")

        ctx = ctypes.c_void_p(None)
        ret = lib.INL_New_Ctx(FIXKEY_CTX, ctypes.byref(ctx))
        if ret != 0:
            err = lib.INL_Error_String(ret)
            err_msg = err.decode("utf-8", errors="replace") if err else "unknown"
            raise RuntimeError(f"INL_New_Ctx failed: code={ret}, {err_msg}")

        if not _atexit_registered:
            atexit.register(lib.INL_Clean)
            _atexit_registered = True
    return ctx

def _get_ctx_a() -> ctypes.c_void_p:
    if not hasattr(_local, "ctx_a"):
        _local.ctx_a = _new_ctx(conf_a)
        if threading.current_thread() is not threading.main_thread():
            weakref.finalize(threading.current_thread(), _free_ctx, _local.ctx_a)
    return _local.ctx_a

def _get_ctx_b() -> ctypes.c_void_p:
    if not hasattr(_local, "ctx_b"):
        _local.ctx_b = _new_ctx(conf_b)
        if threading.current_thread() is not threading.main_thread():
            weakref.finalize(threading.current_thread(), _free_ctx, _local.ctx_b)
    return _local.ctx_b

def encrypt(ctx: ctypes.c_void_p, plaintext: bytes) -> bytes:

    out  = ctypes.c_char_p(None)
    outl = ctypes.c_int(0)
    try:
        ret = lib.INL_Encrypt(ctx, plaintext, len(plaintext),
                              ctypes.byref(out), ctypes.byref(outl))
        if ret != 0:
            raise RuntimeError(f"INL_Encrypt failed: {ret}")

        return out.value[:outl.value]
    finally:
        if out:
            lib.INL_Free_Buf(out)

def decrypt(ctx: ctypes.c_void_p, ciphertext: bytes) -> bytes:

    out  = ctypes.c_char_p(None)
    outl = ctypes.c_int(0)
    try:
        ret = lib.INL_Decrypt(ctx, ciphertext, len(ciphertext),
                              ctypes.byref(out), ctypes.byref(outl))
        if ret != 0:
            raise RuntimeError(f"INL_Decrypt failed: {ret}")

        return out.value[:outl.value]
    finally:
        if out:
            lib.INL_Free_Buf(out)

├── litellm_config.yaml # LiteLLM 프록시 설정
├── my_hook.py          # LiteLLM Hook 로직 (CustomLogger 구현)
└── inisafe.py          # INISAFE Net 초기화 + 암복호화 (예제 및 사용법 참고)

# my_hook.py
from typing import Any, AsyncGenerator, Literal
from litellm.integrations.custom_logger import CustomLogger
from litellm.proxy.proxy_server import UserAPIKeyAuth, DualCache
from litellm.types.utils import ModelResponseStream

from inisafe import encrypt, decrypt  # 예제 및 사용법에서 작성한 inisafe.py


class INISAFENetHook(CustomLogger):

    # LLM 호출 직전
    async def async_pre_call_hook(
        self,
        user_api_key_dict: UserAPIKeyAuth,
        cache: DualCache,
        data: dict,
        call_type: Literal[
            "completion", "text_completion", "embeddings",
            "image_generation", "moderation", "audio_transcription",
        ],
    ):
        # 요청 메시지의 content를 복호화하여 LLM에 평문으로 전달
        for msg in data.get("messages", []):
            if isinstance(msg.get("content"), str):
                msg["content"] = decrypt(msg["content"])
        return data


    # LLM 응답 수신 후
    async def async_post_call_success_hook(
        self,
        data: dict,
        user_api_key_dict: UserAPIKeyAuth,
        response,
    ):
        # LLM 응답 content를 암호화하여 클라이언트에 반환 (non-streaming)
        for choice in getattr(response, "choices", []):
            msg = getattr(choice, "message", None)
            if msg and isinstance(getattr(msg, "content", None), str):
                msg.content = encrypt(msg.content)
        return response


    # SSE 스트리밍 중
    async def async_post_call_streaming_iterator_hook(
        self,
        user_api_key_dict: UserAPIKeyAuth,
        response: Any,
        request_data: dict,
    ) -> AsyncGenerator[ModelResponseStream, None]:
        # 스트리밍 응답의 각 chunk를 암호화하여 클라이언트에 반환
        async for chunk in response:
            for choice in getattr(chunk, "choices", []):
                delta = getattr(choice, "delta", None)
                if delta and isinstance(getattr(delta, "content", None), str):
                    delta.content = encrypt(delta.content)
            yield chunk


proxy_handler_instance = INISAFENetHook()

litellm_settings:
  callbacks: ["my_hook.proxy_handler_instance"]