RAG 22. 권한 기반 검색: 다른 사용자의 문서를 순위 계산 전에 제외하기
답변을 출력할 때만 비공개 문서를 가리면 이미 늦을 수 있습니다. 사용자와 조직 권한을 먼저 확인해, 허용된 본문만 검색 순위 계산 함수에 전달하는 구조를 만듭니다.
답변을 출력할 때만 비공개 문서를 가리면 이미 늦을 수 있습니다. 사용자와 조직 권한을 먼저 확인해, 허용된 본문만 검색 순위 계산 함수에 전달하는 구조를 만듭니다.
그림은 처리 구조를 설명하는 개념도입니다. 실제 실행 화면이나 측정 결과가 아닙니다.
오늘 만들 결과
조직 A의 alice는 공용 정책만 보고, bob은 자신에게 허용된 초안도 봅니다. 조직 B에도 alice라는 이름이 있지만 A의 자료를 볼 수 없습니다. 테스트에서는 최종 결과뿐 아니라 순위 계산 함수가 받은 후보 목록도 검사합니다.
먼저 이해할 것
인증(authentication)은 누가 요청했는지 확인하는 과정이고, 인가 또는 권한 검사(authorization)는 그 사람이 무엇을 볼 수 있는지 결정하는 과정입니다. 테넌트(tenant)는 같은 서비스를 사용하되 데이터를 분리해야 하는 조직 경계입니다.
이번 예제의 Principal은 인증이 끝난 서버 내부 사용자 정보입니다. 생성자 자체가 사용자를 인증하지 않습니다. 실습에서는 직접 만들지만 실제 HTTP 요청의 user_id, tenant_id 값을 그대로 넣으면 안 됩니다. 검증된 로그인 세션이나 토큰에서 서버가 가져와야 합니다.
권한 필터를 상위 결과 뒤에 적용하면 비공개 본문이 재순위화 모델이나 로그에 이미 들어갈 수 있습니다. 그래서 허용된 후보를 데이터 조회 단계에서 선택합니다. 기본값은 거부입니다. 허용 기록이 없는 문서는 보이지 않습니다.
실행 준비와 전체 코드
Python 3.12.14에서 실행했습니다. Python 3.12와 표준 라이브러리만 있으면 됩니다. 빈 실습 폴더를 만들고 아래 파일들을 같은 폴더에 저장하세요. 다른 편의 코드나 유료 API 키는 필요하지 않습니다.
가상의 조직 A·B와 사용자 alice·bob을 코드 안에서 만듭니다. 로그인 시스템을 구현하는 예제가 아니며, 실제 계정이나 비공개 문서는 사용하지 않습니다.
authorized_search.py
import re
import sqlite3
from dataclasses import dataclass
@dataclass(frozen=True)
class Principal:
tenant_id: str
user_id: str
def database():
db = sqlite3.connect(':memory:')
db.executescript("""
CREATE TABLE documents(id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, text TEXT NOT NULL);
CREATE TABLE grants(document_id TEXT NOT NULL, user_id TEXT NOT NULL,
PRIMARY KEY(document_id,user_id));
""")
db.executemany('INSERT INTO documents VALUES(?,?,?)', [
('a-guide', 'A', 'refund policy is 14 days'),
('a-bob', 'A', 'refund private draft for Bob'),
('b-guide', 'B', 'refund policy is 99 days'),
])
db.executemany('INSERT INTO grants VALUES(?,?)', [
('a-guide', 'alice'), ('a-guide', 'bob'), ('a-bob', 'bob'),
('b-guide', 'alice'), # Same username in another tenant.
])
db.commit()
return db
def tokens(text):
return set(re.findall(r'[a-z0-9]+', text.lower()))
def rank(question, candidates, k):
query = tokens(question)
scored = [(len(query & tokens(item['text'])), item) for item in candidates]
scored.sort(key=lambda pair: (-pair[0], pair[1]['id']))
return [item for score, item in scored[:k] if score > 0]
def search(db, principal, question, k=3, ranker=rank):
if not isinstance(principal, Principal) or not principal.tenant_id or not principal.user_id:
raise PermissionError('인증된 사용자 정보가 필요합니다.')
if not isinstance(question, str) or not 1 <= len(question) <= 200:
raise ValueError('질문 길이 오류')
if type(k) is not int or not 1 <= k <= 10:
raise ValueError('k 범위 오류')
rows = db.execute("""
SELECT d.id,d.text FROM documents AS d
WHERE d.tenant_id=? AND EXISTS (
SELECT 1 FROM grants AS g WHERE g.document_id=d.id AND g.user_id=?
) ORDER BY d.id
""", (principal.tenant_id, principal.user_id)).fetchall()
candidates = [dict(id=doc_id, text=text) for doc_id, text in rows]
return ranker(question, candidates, k)
if __name__ == '__main__':
db = database()
for principal in (Principal('A', 'alice'), Principal('A', 'bob'), Principal('B', 'alice')):
print(principal.tenant_id, principal.user_id,
[hit['id'] for hit in search(db, principal, 'refund')])
with db:
db.execute('DELETE FROM grants WHERE document_id=? AND user_id=?', ('a-guide', 'alice'))
print('after revoke:', search(db, Principal('A', 'alice'), 'refund'))
try:
search(db, None, 'refund')
except PermissionError as error:
print('denied:', error)
db.close()
SQL은 tenant_id 일치와 grants의 허용 기록을 동시에 확인합니다. 둘 중 하나만 있으면 부족합니다. 조직 경계를 빼면 다른 조직의 같은 사용자 이름을 잘못 연결할 수 있고, 사용자 허용 기록을 빼면 같은 조직의 모든 문서를 공개하게 됩니다.
rank는 권한이 적용된 candidates만 받습니다. 다른 사용자의 문서가 후보로 들어왔다가 나중에 지워지는 구조가 아닙니다. 외부 재순위화 API를 쓰더라도 이 경계를 유지해야 합니다. 검색 점수와 중간 결과를 남기는 로그에도 권한 없는 본문이 섞이지 않게 해야 합니다.
매 요청마다 허용 기록을 다시 조회하므로 권한 회수 후 다음 검색에서 바로 사라집니다. 실습에는 결과 캐시가 없습니다. 캐시를 붙이면 사용자·조직·권한 버전이 다른 요청 사이에서 결과를 공유하지 않도록 설계해야 합니다. 응답 생성 중 권한이 바뀌는 상황도 필요한 위협 모델에 따라 다시 검사하세요.
이 작은 SQLite 예제는 애플리케이션 필터의 위치를 보여 줍니다. 서비스 계정의 데이터베이스 권한, 원문 다운로드 API, 관리자 접근, 동시 요청까지 검증한 완전한 멀티테넌트 제품은 아닙니다.
직접 실행한 결과
python authorized_search.py
아래는 이 글의 가상 데이터로 로컬에서 실제 실행한 출력입니다. 언어 모델이나 임베딩 API를 호출한 결과가 아닙니다.
A alice ['a-guide']
A bob ['a-bob', 'a-guide']
B alice ['b-guide']
after revoke: []
denied: 인증된 사용자 정보가 필요합니다.
테스트로 경계를 고정하기
다음을 test_example.py로 저장합니다.
import unittest
from authorized_search import database, Principal, search, rank
class AuthorizationTests(unittest.TestCase):
def setUp(self):
self.db = database()
def tearDown(self):
self.db.close()
def test_unauthorized_text_never_reaches_ranker(self):
seen = []
def spy(question, candidates, k):
seen.extend(candidates)
return rank(question, candidates, k)
hits = search(self.db, Principal('A', 'alice'), 'refund', ranker=spy)
self.assertEqual([row['id'] for row in seen], ['a-guide'])
self.assertEqual([row['id'] for row in hits], ['a-guide'])
def test_tenant_boundary_with_same_user_name(self):
hits = search(self.db, Principal('B', 'alice'), 'refund')
self.assertEqual([row['id'] for row in hits], ['b-guide'])
def test_no_grant_and_missing_identity_deny(self):
self.assertEqual(search(self.db, Principal('A', 'unknown'), 'refund'), [])
with self.assertRaises(PermissionError):
search(self.db, None, 'refund')
def test_revoke_takes_effect(self):
self.db.execute("DELETE FROM grants WHERE user_id='alice'")
self.db.commit()
self.assertEqual(search(self.db, Principal('A', 'alice'), 'refund'), [])
def test_user_id_is_parameterized(self):
self.assertEqual(search(self.db, Principal('A', "alice' OR 1=1 --"), 'refund'), [])
if __name__ == '__main__':
unittest.main()
python -m unittest -v
다섯 테스트가 순위 계산 전 후보 제한, 조직 분리, 기본 거부, 권한 회수, SQL 값 바인딩을 확인합니다. 인증 토큰 검증은 구현·테스트 범위에 없습니다.
test_no_grant_and_missing_identity_deny (test_example.AuthorizationTests.test_no_grant_and_missing_identity_deny) ... ok
test_revoke_takes_effect (test_example.AuthorizationTests.test_revoke_takes_effect) ... ok
test_tenant_boundary_with_same_user_name (test_example.AuthorizationTests.test_tenant_boundary_with_same_user_name) ... ok
test_unauthorized_text_never_reaches_ranker (test_example.AuthorizationTests.test_unauthorized_text_never_reaches_ranker) ... ok
test_user_id_is_parameterized (test_example.AuthorizationTests.test_user_id_is_parameterized) ... ok
----------------------------------------------------------------------
Ran 5 tests in 0.001s
OK
직접 바꿔 볼 것
alice에게a-bob허용을 추가했다가 회수하세요. 다음 검색 결과가 즉시 바뀌는지 검사하세요.- 다른 문서 다운로드 함수가 있다면 같은 권한 검사를 붙이세요. 검색만 막고 원문 링크가 열리면 경계가 깨집니다.
- 결과 캐시를 추가하기 전에,
alice의 결과가bob에게 재사용되지 않는 테스트부터 작성하세요. - 벡터 저장소로 바꿀 때에는 지원되는 서버 측 메타데이터 필터를 확인하고, 후보를 외부로 보낸 뒤 거르는 구조를 피하세요.
참고 자료
- OWASP Authorization Cheat Sheet: 최소 권한, 기본 거부, 매 요청의 권한 검증 원칙.
RAG 실전 포트폴리오 22/25
이전: 21. 문서 갱신: 같은 내용은 건너뛰고 오래된 청크는 지우자