Ficket은 얼굴 인식을 기반으로 한 티켓팅 서비스로, 사용자가 등록한 얼굴 이미지를 통해 현장에서 티켓 소지 여부를 확인할 수 있는 시스템입니다. 이 시스템은 InsightFace를 활용해 얼굴 임베딩(특징 벡터)을 생성하고, AES 암호화를 통해 데이터를 DB에 관리하며, AWS S3를 사용해 이미지를 저장합니다.
설치
1) 라이브러리 설치
다음 명령어를 실행하여 프로젝트에 필요한 라이브러리를 설치합니다.
C++이 없으면 오류가 발생하니 설치 필요합니다.
pip install insightface flask flask-restx opencv-python-headless numpy pycryptodome boto3 apscheduler2) s3 설정 및 sse-kms 적용
2) S3 및 암호 키, KMS 설정
face-service-local.yml 파일에 S3 정보와 벡터값을 AES 암호화를 위한 secret key, S3-KMS ARN 값을 설정합니다.
flask:
mysql:
url: mysql+pymysql://root:{password}@localhost:3306/msa-exercise?charset=utf8
password: "{cipher}AQBzuG1fOvuKxvTC1TXksA9FVwIHjHkduMGtKFzru6DkuWMyegUWEmEabfgvjIf3wkE3iepKtjWsaCfh+5tGJnmlJtqwL5zqYRSHufLxs84MoqjBE9yV9lbnrpvr7/PboqdQ/Jn4792QpjESos6h6bcUqGVMksP5PclqfYFPWf9/dZBfOdeVmWMm8n+hyhBnSJO0+XetpEyrO2/znrZoer/10iwCZ5WtBfoTuAnRhxcY94Ujb5LZ0xEE0CwBl2jMbs/uhwhqDJhXDd3RVZ3UHFprt7SqdZCKZRZkr6DlVsLRppoe+1YBFbHleKjGlkT6YDECXYNnCSAfhfu5d5spMkm/iVqbY1ScKVQ/8vu59V8cAzuwO9ensILhA/H3nvg0urk="
encryption:
secret_key: "{cipher}AQCPh+nVMl+4uuQJsr0oGmAu0z40eNSM9JoVwvFica+M9B2RfHx/keku4ooCPpHOtxXDXT5m6w/dF+hgPXteIctVUjLwaqNb3NR8gL+gF9s+UP2I2qWzo1QG+xT/OatCSgp3x530yYTZ4xKTit1VffFIKBI5dJ7rOMs73eQX6Ql4yeCy344Ib+grp3W9EJF8zvpXfzz2kjd7Xf9SHSKo4k6+JrVapPy/GDN8HliIDnD075GZho+ecRm5s60i3EcVm8dxodW+9PHqpFFmaH0hpYZkf56i8HIt3yE+3yQn61HBC7j1STZh73uOgnl6hyejgiG1imtXFO7x1GBOMwl6J/oxRlkGK61SdL8ZBBLyc09fFdFYC+EMlgvWHO3rT7dh3H9wJazFIHak+LGUve/cakwJ"
s3:
aws:
accesskey: "{cipher}AQBBpTDKAfl0aXPYmPzSfphzKw2baCH4WooqgsGidwc9ypAcOqqIB03v3qpoqrxYuZkA4fD/7NiYNvLizgK5J85O/ILHYwtTPVbunH4D4qDo8gJjz18UO02V/shE1N+qyNXVAEm/5uGA0UiOqf8BsZtnIAFinSSImPd7vOFzBf8gGgbrBDMniPNH/QdN+lmwCXCv1KtbiZUQ2QndkUn93AotgdClQopxzysX7FFxleqJjddooFm6TWRKUURR7FBRMwS+x8h1m0PpwrjyZYgxjv+gmnxXCIUQNX/qVec/TA6KLOYzw8ePqPx0kDveoQuFLzEtoO3CcgKIQQHpIR+ZxuX7z52S7HSoqtDq2tyDVJTrSAF4PnJAZVw7G1BR87VFbMhqxgMey58o7C0FVh3oSnxD"
secretkey: "{cipher}AQB8X7MBmqVQpnMRqJCK9o2FDcgOnstV8GzL8gbqk+l2a1UPnQVPKjdUVvzpN7IbLxlSMY1Tac/+CzAEfnHaF0uJUkV0xO/SxlvYvYGg3XUT9XCLjosb3AiulFRuADarWChHDIc4XdHDPoPvvRqS+ui5pfi+ySN5HUjkr+R/cn+5b8xE7xXeSespZeiH/iJOnXWkwmVVQN6pc3UCDkQKmIVZly2mfuynJJvpsAKnY9QBalsAsDt8S+ycw2kRjSsZryK/Qx/U9swFN8TBQMgxCkA8HxpODAwV+CsV0WSceY5wbUR7gbzvoO9P35u5hxgy8kP5q6XbDtx+0H3PjK6W3BM3KhU0uqJGHdt5e9dFzcQnT2b+JsBomoBSQZhTpvCde1pyWft9jreLT3185QwPRFURpELBvfAL2wY3Or2OgThQZw=="
bucketname: ficket-event-content
region: ap-northeast-2
kms: "{cipher}AQB998ZzYPTlUIQSU/7/TSYldmrSc7VfnGl3EAlTPNswxluUvub0dYG0wmRostcrbjrIstUzbPoT5cuRWq78S2AFEt6uNinWDsnHUdeu3wzh72ETvbiBS4tlgRSANFzQZ6pc5Eq7zruHoRjv7F9oc0+ukmSuMBalCtu+FK9WoDgcBO4p7+LoirnIO9/ipdWH54IJwB25AUbCK+SSWtYbgrzQg81akBlBk9U63DE2PENoT2tfF5G/lnpe3sep2007bTWeeH0OYDhnh+9o8jWcwKLdsi+CtrmtOAGhdYPfS3R0iELNZ3sMHr+g8FmeQndEvd1MpskVHv00BujtcUEgyHky9L5q7wdpgo5okCWpo26/Y9i8CR3ocb0bPqwS6w/gsP55OMbNmdLQP42atLrGkoX5+kAExPO8OFnuq3A6EWJSGqOBiktyxLHUru3+U++zOmpzDIk1xl3ZFP/z+DMkIESrOeG7wwFHn90i3gVD7WcK0Q=="
주요 기능
1) S3 파일 업로드
upload_file_to_s3 함수는 파일을 고유한 이름으로 저장하고, S3 URL을 반환합니다.
import uuid
from config.s3_config import s3, bucket_name
def upload_file_to_s3(file, folder="faces"):
file_extension = file.filename.split('.')[-1]
unique_filename = f"{folder}/{uuid.uuid4().hex}.{file_extension}"
s3.upload_fileobj(
file,
bucket_name,
unique_filename,
ExtraArgs={"ContentType": file.content_type}
)
file_url = f"https://{bucket_name}.s3.amazonaws.com/{unique_filename}"
return file_url2) 얼굴 임베딩 생성
InsightFace의 FaceAnalysis 객체를 사용해 업로드된 얼굴 이미지에서 임베딩(벡터)을 생성합니다.
from insightface.app import FaceAnalysis
import cv2
import numpy as np
app = FaceAnalysis(name='buffalo_l', allowed_modules=['detection', 'recognition'], providers=['CPUExecutionProvider'])
app.prepare(ctx_id=-1)
def get_face_embedding(image_data):
img_array = np.frombuffer(image_data, np.uint8)
img = cv2.imdecode(img_array, cv2.IMREAD_COLOR)
img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
faces = app.get(img)
if faces:
return faces[0].embedding
else:
print("No face detected in the provided image data")
return None3) 벡터 암호화 및 복호화
AES CBC 모드를 사용해 얼굴 임베딩 데이터를 안전하게 암호화 및 복호화합니다.
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
def encrypt_vector(embedding, secret_key):
data = embedding.tobytes()
cipher = AES.new(secret_key, AES.MODE_CBC)
ct_bytes = cipher.encrypt(pad(data, AES.block_size))
iv = base64.b64encode(cipher.iv).decode('utf-8')
ct = base64.b64encode(ct_bytes).decode('utf-8')
return iv + ct
def decrypt_vector(encrypted_data, secret_key):
iv = base64.b64decode(encrypted_data[:24])
ct = base64.b64decode(encrypted_data[24:])
cipher = AES.new(secret_key, AES.MODE_CBC, iv)
decrypted_data = unpad(cipher.decrypt(ct), AES.block_size)
return np.frombuffer(decrypted_data, dtype=np.float32)API
1) 얼굴 등록 API (/upload)
얼굴 이미지를 분석하여 생성된 임베딩 데이터를 암호화하여 DB에 저장하고, S3에 이미지를 저장합니다.
@api.route("/upload")
class UploadFace(Resource):
@api.expect(file_upload_parser)
def post(self):
args = file_upload_parser.parse_args()
file = request.files.get("file")
event_schedule_id = args.get("event_schedule_id")
if not file:
return ResponseSchema.make_response(400, "File not provided.")
image_data = file.read()
embedding = get_face_embedding(image_data)
if embedding is None:
return ResponseSchema.make_response(400, "No face detected.")
encrypted_embedding = encrypt_vector(embedding, secret_key)
file.seek(0)
file_url = upload_file_to_s3(file)
new_face = Face(
vector=encrypted_embedding,
face_img=file_url,
ticket_id=None,
event_schedule_id=event_schedule_id
)
db.session.add(new_face)
db.session.commit()
return ResponseSchema.make_response(200, "Face uploaded successfully.", {
"faceId": new_face.face_id,
"faceUrl": new_face.face_img
})2) 얼굴 매칭 API (/match)
업로드된 이미지를 벡터화하고 데이터베이스에 저장된 암호화된 벡터를 복호화하여 매칭 여부를 확인합니다.
@api.route("/match")
class MatchFace(Resource):
@api.expect(match_parser)
def post(self):
args = match_parser.parse_args()
file = request.files.get("file")
event_schedule_id = args.get("event_schedule_id")
if not file or not event_schedule_id:
return ResponseSchema.make_response(400, "File or event_schedule_id missing.")
image_data = file.read()
embedding = get_face_embedding(image_data)
if embedding is None:
return ResponseSchema.make_response(400, "No face detected.")
faces = Face.query.filter_by(event_schedule_id=event_schedule_id).all()
if not faces:
return ResponseSchema.make_response(404, "No faces found for the event schedule.")
max_similarity = -1
best_match = None
for face in faces:
decrypted_embedding = decrypt_vector(face.vector, secret_key)
similarity = cosine_similarity(embedding, decrypted_embedding)
if similarity > max_similarity:
max_similarity = similarity
best_match = {
"face_id": face.face_id,
"face_img": face.face_img,
"ticket_id": face.ticket_id,
"event_schedule_id": face.event_schedule_id,
"similarity": float(similarity),
}
threshold = 0.4
if best_match and max_similarity > threshold:
return ResponseSchema.make_response(200, "Face match found.", best_match)
else:
return ResponseSchema.make_response(404, "No matching face found.")Reference
