I Love Cloud Computing, 안녕하세요. ManVSCloud의 IT 기술 블로그입니다.

NCLOUD

[NCLOUD] Key Management Service를 활용한 데이터베이스 암호화 및 복호화 가이드

안녕하세요. MANVSCLOUD 김수현입니다.

최근에 암호화 알고리즘에 대한 포스팅을 작성했었는데요. 암호화하면 데이터 암호화와 관련하여 키 관리 서비스를 빼놓을 수 없을 것같습니다.
그러므로 오늘은 Key Management Service에 대한 주제로 포스팅을 진행하겠습니다.

Key Management Service

Key Management Service(KMS)는 암호화 키를 생성, 저장, 관리, 키 회전 및 키 폐기와 같은 키 수명주기 관리를 담당하는 서비스입니다. 이를 통해 사용자는 민감한 데이터를 안전하게 암호화하고 복호화할 수 있으며, 권한이 있는 사용자만 해당 키에 액세스할 수 있습니다.

※ 주요 기능

강력한 암호화 키 생성
안전한 키 저장소에 암호화 키를 보관
권한이 있는 사용자 및 응용 프로그램만 사용할 수 있도록 암호화 키 액세스를 제어
정기적으로 키를 자동으로 교체하여 보안 유지
필요하지 않은 키를 안전하게 폐기

암호화 키 관리가 중요한 데이터 보안 요소 중 하나인 만큼, Key Management Service(KMS)는 기업 및 개인의 데이터를 보호하는 데 필수적인 도구입니다.
KMS를 사용하면 암호화 키의 생성, 저장, 관리 및 폐기와 같은 중요한 작업을 안전하고 효율적으로 수행할 수 있기때문에 보안 요구 사항에 따라 데이터 보안을 강화하고 감사 요구 사항 충족 및 관리의 편의성을 높일 수 있습니다.

NAVER CLOUD PLATFORM – KMS

네이버 클라우드 플랫폼에서도 관리형 Key Management Service가 존재합니다.
관리형 서비스를 사용하지 않고 직접 Vault, OpenKeychain, EJBCA 등을 오픈 소스를 활용하여 특정 요구 사항에 맞는 사용자 정의 KMS 솔루션을 구축할 수도 있으나 이를 위해서는 암호화 및 키 관리에 대한 깊은 이해와 안전한 키 저장, 키 생성, 키 회전 및 키 폐기와 같은 기능을 구현하는 데 필요한 전문 지식이 필요합니다.

그렇다면 KMS는 어떤 사례에서 어떻게 사용할 수 있을까요?
아마 많은 사용자들이 현재 KMS 사용 가이드만 보고 개념은 터득했지만 어떻게 사용해야할지 몰라서 사용하지 못하는 경우가 많을 것이라 생각됩니다.

KMS를 사용하는 몇 가지 일반적인 사례는 다음과 같습니다.

파일 저장소 암호화 : Object Storage 또는 로컬 파일 시스템에서 저장되는 파일을 암호화할 때 KMS를 사용할 수 있습니다.
애플리케이션 레벨 암호화 : 사용자 데이터, 메시지, 또는 특정 애플리케이션에 필요한 기타 민감 정보를 암호화하는 데 KMS를 사용할 수 있습니다. 이를 통해 데이터를 보호하고, 데이터 유출 또는 무단 접근 시도를 방지할 수 있습니다.
IoT 및 디바이스 보안: IoT 디바이스 및 기타 연결된 시스템에서 사용되는 암호화 키를 안전하게 생성, 저장 및 관리하기 위해 KMS를 사용할 수 있습니다.
디지털 서명 및 인증: KMS를 사용하여 디지털 서명 및 인증에 사용되는 키를 안전하게 생성 및 저장할 수 있습니다. 이를 통해 전자 문서, 이메일, 소프트웨어 패키지 등의 무결성 및 출처를 검증할 수 있습니다.

그 외에도 다양한 방향으로 KMS 서비스를 활용할 수 있는데요. 예를 들어 아래와 같은 사례를 예시로 들어봅시다.

병원에서 환자의 개인정보를 안전하게 보호하려면 어떻게 해야 할까요?
병원은 환자의 민감한 개인정보와 의료 데이터를 다루기 때문에 데이터 보안에 특히 신경을 써야 합니다. 데이터베이스 내의 환자 정보를 보호하는 방법으로 Key Management Service(KMS)를 사용하여 데이터를 암호화하는 것은 효과적인 방법이 될 수 있습니다.

KMS를 사용하여 데이터베이스에 저장되는 모든 환자 데이터를 암호화하고 의사가 환자의 정보를 조회할 때 시스템은 환자 데이터를 복호화하고 의사에게 보여주도록 사용할 수 있겠죠.

이제 네이버 클라우드 플랫폼의 Key Management Service를 직접 사용해보도록 하겠습니다.

Data Encryption/Decryption

먼저 Security – Key Management Service 에서 [+ 키 생성]을 이용하여 키를 생성하도록 하겠습니다.

생성된 키는 아래 그림과 같이 Object Storage를 생성할 때도 KMS 마스터 키를 설정하여 사용할 수 있습니다.

하지만 KMS가 오직 Object Storage나 Private CA와 같은 서비스와 연동하여 사용되는 것을 목적으로 만들어진 것은 아닙니다. 그러므로 Key Management Service의 API를 이용하여 활용 범위를 넓혀보도록 하겠습니다.

KMS 개요

KMS API는 RESTful 형태로 제공됩니다. KMS Rest API를 호출하기 위해서는 API Gateway와 Sub Account 인증을 수행해야 합니다.

아래 코드는 암/복호화를 간단하게 테스트 해볼 수 있는 테스트 코드입니다.
ACCESS_KEY, SECRET_KEY, KMS_KEY_TAG와 같이 민감한 정보는 보안상 환경 변수나 별도의 설정 파일을 활용하여 관리가 필요합니다.

🔒 Encrypt

import hashlib
import hmac
import base64
import requests
import time
import json
from datetime import datetime

def main():

    timestamp = int(time.time() * 1000)
    timestamp = str(timestamp)

    access_key = "{ACCESS_KEY}"
    secret_key = "{SECRET_KEY}"
    secret_key = bytes(secret_key, 'UTF-8')

    method = "POST"
    key_tag = "{KMS_KEY_TAG}"

    api_server = "https://kms.apigw.ntruss.com"
    uri = "/keys/v2/" + key_tag + "/encrypt"
    uri = uri + "?responseFormatType=json"

    message = method + " " + uri + "\n" + timestamp + "\n" + access_key
    message = bytes(message, 'UTF-8')
    signingKey = base64.b64encode(hmac.new(secret_key, message, digestmod=hashlib.sha256).digest())

    http_header = {
        'Content-Type': 'application/json',
        'x-ncp-apigw-signature-v2': signingKey,
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key
        }

    # 암호화할 텍스트 입력 및 Base64 형태로 인코딩
    text_to_encrypt = 'MANVSCLOUD'
    text_to_encrypt_base64 = base64.b64encode(text_to_encrypt.encode('utf-8')).decode('utf-8')

    payload = {
        "plaintext": text_to_encrypt_base64
    }

    response = requests.post(api_server + uri, headers=http_header, json=payload)

    data = json.loads(response.text)
    ciphertext = data['data']['ciphertext']

    print(ciphertext)

if __name__ == '__main__':
    main()

🔓 Decrypt

import hashlib
import hmac
import base64
import requests
import time
import json
from datetime import datetime

def main():

    timestamp = int(time.time() * 1000)
    timestamp = str(timestamp)

    access_key = "{ACCESS_KEY}"
    secret_key = "{SECRET_KEY}"
    secret_key = bytes(secret_key, 'UTF-8')

    method = "POST"
    key_tag = "{KMS_KEY_TAG}"

    api_server = "https://kms.apigw.ntruss.com"
    uri = "/keys/v2/" + key_tag + "/decrypt"
    uri = uri + "?responseFormatType=json"

    message = method + " " + uri + "\n" + timestamp + "\n" + access_key
    message = bytes(message, 'UTF-8')
    signingKey = base64.b64encode(hmac.new(secret_key, message, digestmod=hashlib.sha256).digest())

    http_header = {
        'Content-Type': 'application/json',
        'x-ncp-apigw-signature-v2': signingKey,
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key
        }


    payload = {
        "ciphertext": "{CIPHERTEXT}"
    }

    response = requests.post(api_server + uri, headers=http_header, json=payload)

    data = json.loads(response.text)
    origintext = data['data']['plaintext']

    encoded_string = origintext
    decoded_bytes = base64.b64decode(encoded_string)
    decoded_string = decoded_bytes.decode("utf-8")

    print(decoded_string)


if __name__ == '__main__':
    main()

ACCESS_KEY : Sub Account에서 추가한 Access Key Id (Main 계정의 인증키로도 사용가능하지만 권장하지 않음)
SECRET_KEY : Sub Account에서 생성된 Access Key의 Secret Key
KMS_KEY_TAG : Key Management Service 콘솔에서 생성 후 키를 선택하여 ‘키 Tag’ 확인 가능
CIPHERTEXT : KMS를 이용해 평문을 암호화한 암호문

Key Management Service를 이용하여 키를 발급하고 위 테스트 코드를 통해 충분히 암/복호화를 진행해보았다면 이제 위에서 사용된 API를 활용하여 데이터베이스에 저장되는 데이터를 KMS를 활용하여 암호화하고 웹에서 보여지는 부분은 복호화하여 보여질 수 있도록 해보겠습니다.

A to Z

# VPC, Server 생성에 대한 가이드는 생략합니다.
# ACG 설정은 (Inbound) 22, 10000 / (Outbound) 80, 443이 허용되었습니다.
# Python은 3.7.13 버전이 사용되었습니다.
# MySQL 8버전이 사용되었습니다.
# 해당 A to Z는 Python 개발 환경 구성이 완료되었다는 가정하에 진행됩니다.

해당 A to Z 과정은 Flask와 MySQL을 이용하여 진행합니다.
다만 Flask, MySQL 설치 및 설정에 대한 내용은 주제에서 벗어나므로 세부적인 설명은 생략합니다.

MySQL이 설치되었다는 가정하에 DB 접속 후 아래와 같이 계정 및 테이블을 생성해주었습니다.

mysql> use mysql;
mysql> CREATE DATABASE manvscloud default CHARACTER SET UTF8;
mysql> create user 'manvscloud'@'localhost' identified by 'PASSWORD';
mysql> grant all privileges on manvscloud.* to 'manvscloud'@'localhost'
mysql> flush privileges;
mysql> use manvscloud;
mysql> CREATE TABLE mvsc (
    id INT AUTO_INCREMENT PRIMARY KEY,
    secret VARCHAR(255) NOT NULL
);

mysql> describe mvsc;
+--------+--------------+------+-----+---------+----------------+
| Field  | Type         | Null | Key | Default | Extra          |
+--------+--------------+------+-----+---------+----------------+
| id     | int          | NO   | PRI | NULL    | auto_increment |
| secret | varchar(255) | NO   |     | NULL    |                |
+--------+--------------+------+-----+---------+----------------+
2 rows in set (0.00 sec)

Flask와 MySQL을 연동하기 위해 flask-mysqldb도 함께 설치해주었습니다.

pip install requests flask flask-mysqldb

Flask와 MySQL이 잘 연동되었는지 확인과 함께 기본적으로 테스트할 환경을 구성하고 웹 사이트에 접속하여 값을 입력하고 조회까지 할 수 있도록 해보겠습니다.

app.py

아래 코드에서 MySQL의 User, Password, DB이름, 테이블 명은 설정한대로 변경해주자!

from flask import Flask, render_template, request, redirect, url_for
from flask_mysqldb import MySQL

app = Flask(__name__)

# 데이터베이스 설정
app.config['MYSQL_HOST'] = 'localhost'
app.config['MYSQL_USER'] = 'manvscloud'
app.config['MYSQL_PASSWORD'] = 'PASSWORD'
app.config['MYSQL_DB'] = 'manvscloud'

mysql = MySQL(app)

@app.route('/')
def index():
    cur = mysql.connection.cursor()
    cur.execute("SELECT * FROM mvsc")
    data = cur.fetchall()
    cur.close()
    return render_template('index.html', data=data)

@app.route('/submit', methods=['POST'])
def submit():
    if request.method == 'POST':
        user_input = request.form['data']
        cur = mysql.connection.cursor()
        cur.execute("INSERT INTO mvsc (secret) VALUES (%s)", (user_input,))
        mysql.connection.commit()
        cur.close()
        return redirect(url_for('index'))

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=10000)

templates/index.html

<!DOCTYPE html>
<html lang="ko">
<head>
    <meta charset="UTF-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Key Management Service를 이용한 DB 데이터 암/복호화</title>
</head>
<body>
    <h1>DB에 저장할 데이터 입력</h1>
    <form action="{{ url_for('submit') }}" method="post">
        <label for="data">데이터:</label>
        <input type="text" id="data" name="data" required>
        <button type="submit">Submit</button>
    </form>

    <h2>저장된 데이터 목록</h2>
    <ul>
        {% for row in data %}
            <li>{{ row[1] }}</li>
        {% endfor %}
    </ul>
</body>
</html>

코드가 작성되었다면 python app.py로 앱을 실행시켜 웹에 접속해봅니다.
데이터 : 부분에 “테스트입니다.”라고 입력 후 [Submit] 버튼을 클릭하게되면 데이터베이스에 저장된 “테스트입니다.”라는 값이 리스트 형태로 조회되어 나옵니다.

mysql> delete from mvsc where id = 1;
Query OK, 1 row affected (0.01 sec)

이제 테스트로 입력한 값을 삭제해주고 데이터 입력 시 KMS를 이용하여 암호화되고 조회 시에는 복호화하여 볼 수 있도록 코드를 변경해봅시다.

변경된 app.py

import hashlib
import hmac
import base64
import requests
import time
import json
from datetime import datetime
from flask import Flask, render_template, request, redirect, url_for
from flask_mysqldb import MySQL

app = Flask(__name__)

# 데이터베이스 설정
app.config['MYSQL_HOST'] = 'localhost'
app.config['MYSQL_USER'] = 'manvscloud'
app.config['MYSQL_PASSWORD'] = 'PASSWORD'
app.config['MYSQL_DB'] = 'manvscloud'

mysql = MySQL(app)

def get_api_config():
    access_key = "{ACCESS_KEY}"
    secret_key = "{SECRET_KEY}"
    secret_key = bytes(secret_key, 'UTF-8')
    key_tag = "{KMS_KEY_TAG}"
    api_server = "https://kms.apigw.ntruss.com"

    return access_key, secret_key, key_tag, api_server


# 복호화 함수
def decrypt_text(ciphertext):

    access_key, secret_key, key_tag, api_server = get_api_config()

    timestamp = int(time.time() * 1000)
    timestamp = str(timestamp)

    method = "POST"

    uri = "/keys/v2/" + key_tag + "/decrypt"
    uri = uri + "?responseFormatType=json"

    message = method + " " + uri + "\n" + timestamp + "\n" + access_key
    message = bytes(message, 'UTF-8')
    signingKey = base64.b64encode(hmac.new(secret_key, message, digestmod=hashlib.sha256).digest())

    http_header = {
        'Content-Type': 'application/json',
        'x-ncp-apigw-signature-v2': signingKey,
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key
        }

    payload = {
        "ciphertext": ciphertext
    }

    response = requests.post(api_server + uri, headers=http_header, json=payload)
    data = json.loads(response.text)

    if 'data' not in data or 'plaintext' not in data['data']:
        print("Error: Unable to retrieve the plaintext from the API response.")
        print(data)
        return None

    origintext = data['data']['plaintext']
    decoded_bytes = base64.b64decode(origintext)
    decoded_string = decoded_bytes.decode("utf-8")

    return decoded_string


# 암호화 함수
def encrypt_text(plaintext):

    access_key, secret_key, key_tag, api_server = get_api_config()

    timestamp = int(time.time() * 1000)
    timestamp = str(timestamp)

    method = "POST"

    uri = "/keys/v2/" + key_tag + "/encrypt"
    uri = uri + "?responseFormatType=json"

    message = method + " " + uri + "\n" + timestamp + "\n" + access_key
    message = bytes(message, 'UTF-8')
    signingKey = base64.b64encode(hmac.new(secret_key, message, digestmod=hashlib.sha256).digest())

    http_header = {
        'Content-Type': 'application/json',
        'x-ncp-apigw-signature-v2': signingKey,
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key
        }

    text_to_encrypt_base64 = base64.b64encode(plaintext.encode('utf-8')).decode('utf-8')

    payload = {
        "plaintext": text_to_encrypt_base64
    }

    response = requests.post(api_server + uri, headers=http_header, json=payload)
    data = json.loads(response.text)

    ciphertext = data['data']['ciphertext']
    return ciphertext


# 메인 페이지
@app.route('/')
def index():

    cur = mysql.connection.cursor()
    cur.execute("SELECT * FROM mvsc")
    data = cur.fetchall()
    cur.close()

    # 데이터 복호화
    decrypted_data = []
    for row in data:
        decrypted_text = decrypt_text(row[1])
        if decrypted_text is not None:
            decrypted_data.append((row[0], decrypted_text))

    return render_template('index.html', data=decrypted_data)

# 데이터 SUBMIT
@app.route('/submit', methods=['POST'])
def submit():
    if request.method == 'POST':
        user_input = request.form['data']
        encrypted_data = encrypt_text(plaintext=user_input) 
        cur = mysql.connection.cursor()
        cur.execute("INSERT INTO mvsc (secret) VALUES (%s)", (encrypted_data,))
        mysql.connection.commit()
        cur.close()
        return redirect(url_for('index'))


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=10000)

기존 코드에서 KMS API로 암호화 및 복호화하는 코드가 추가되었습니다.
이제 다시 python app.py로 앱을 실행하여 데이터를 입력해보고 결과를 확인해봅시다.

웹에서 데이터 입력 시 데이터는 암호화되어 데이터베이스에 저장됩니다.
하지만 웹에서 조회되는 값은 복호화된 평문 값을 조회할 수 있습니다.

여기까지 잘 따라오셨다면 Key Management Service의 API를 이용하여 데이터를 암호화하고 복호화하는 방법을 배우셨을 것입니다.

이제 이러한 방법을 다양한 측면으로 고려하여 서비스에 통합해보시길 바랍니다.

Personal Comments

보안은 현대의 애플리케이션 개발에서 매우 중요합니다.데이터 보안은 사용자들의 신뢰를 얻고 비즈니스에 긍정적인 영향을 미치는 데 결정적인 요소가 됩니다.

이 글을 통해 데이터를 암호화 및 복호화 해보았고 클라우드 기반의 관리형 Key Management Service를 사용하여 암호화 키를 엄격하고 안전하게 보호하며 편리하게 사용할 수 있는 것을 알 수 있었습니다.

긴 글 읽어주셔서 감사합니다.

4월 15, 2023 by manvscloud

Database