[기타] NCP Object Storage + Global CDN 으로 안전하게 미디어파일 제공하기
이미지, 음원, 동영상등과 같은 정적 파일이나 미디어파일을 제공하기 위해서는 다양한 방법을 고려해볼 수 있습니다. 이번 포스트에서는 원본 파일에 대한 보안이 필요한 경우 활용할 수 있는 방법에 대하여 알아보겠습니다.
원본 파일 보안
인터넷을 통해 제공되는 정적 파일이나 미디어 파일은 사용자가 접근이 가능한 경우 손쉽게 다운로드도 가능합니다. DRM 솔루션을 도입하면 원본 파일이 다운로드 되더라도 정상적인 이용이 불가능하겠지만 비용이 발생하거나 기술적 구현 난이도가 높다는 단점이 있습니다.
제가 소속되어있는 개발 조직에서는 음원 파일을 제공하는 오디오북 서비스를 운영중입니다. 기존의 음원 제공 방식에서는 사용자가 간단한 조작만으로도 여러 음원에 접근이 가능하였기에 원본에 대한 보안 처리가 필요하게 되었습니다.
NCP Object Storage 원본 파일 관리
우선 원본 파일에 대해 저장하고 관리할 공간이 필요합니다. NAS나 on-premise 환경에 저장 공간이 여유롭다면 이를 사용하면 되겠으나 외부에서 접근 시 권한 확인을 위한 별도의 기능을 추가적으로 구현해야합니다.
제가 소속되어있는 개발 조직은 서비스 운영을 Naver Cloud Platform(이하 NCP)환경을 이용하고 있었기 때문에 Object Storage에 원본 파일을 저장하였습니다. 이번 포스트의 예시도 Object Storage에서 제공하는 기능을 사용하여 원본에 대한 접근을 제한해보겠습니다.
Object Storage 버킷 생성
NCP에 접속하여 원본 파일을 저장할 버킷을 생성해보겠습니다.
버킷 이름을 입력하고 기본 설정을 유지한 채 3.권한 관리로 넘어갑니다.
권한 관리 화면에서 전체 공개 여부를 공개로 설정합니다. 개별 파일에 대한 공개 권한이 아닌 버킷 내 폴더/디렉토리 리스트의 공개 여부를 설정하는 값으로 공개 안함으로 설정 할 겨우 개별 파일에 대한 권한을 공개하여도 접근하지 못하는 문제가 발생합니다.
파일 업로드 및 접근 테스트
접근 테스트에 사용할 샘플 음원 파일을 버킷에 업로드하였습니다.
권한 관리부분을 보시면 공개로 설정되어있습니다. 이제 상세 정보의 Link 정보로 음원 파일에 접근해보겠습니다.
간단하게 원본 파일에 접근이 가능합니다. 사용자는 이러한 방법으로 원본을 자유롭게 이용할 수도, 다운로드 할 수도 있습니다. 이제 공개 여부를 공개 안함으로 설정 후 동일한 요청을 해보겠습니다.
권한을 공개 안함으로 설정하니 원본 파일에 접근할 수 없게 되었습니다. 기본적으로 원본 파일들은 공개 안함상태로 업로드가 될 것입니다.(NCP Amazon S3 SDK 가이드) 공개되지 않은 파일에 접근하는 방법으로 Presigned URL 방식과 Global CDN(Secure Token) 방식이 있습니다. 이중 Object Storage를 Global CDN과 연동하여 Secure Token을 발급하는 방식에 대하여 알아보도록 하겠습니다.
Global CDN 생성
NCP에서는 CDN+라는 이름으로 CDN서비스를 제공하고있었으나 deprecate이후 Global CDN 서비스를 출시하였습니다. NCP console에 접속하여 새로운 Global CDN을 생성해보겠습니다.
0.생성: +CDN 신청 버튼을 클릭하여 생성 화면으로 이동하겠습니다.
1.서비스 설정: CDN이름을 입력하고 서비스 프로토콜을 ALL로 설정하였습니다.(프로토콜 부분은 사용자 요구사항에 따라 변경하셔도 됩니다. 예제임을 감안하여 ALL로 설정하겠습니다.)
2.원본 설정: 해당 메뉴에서는 어떠한 버킷의 원본을 CDN에서 사용할지에 대한 설정화면입니다. 이전에 생성한 cdn-test-bucket을 원본 위치로 설정합니다. 특정 디렉토리의 파일을 원본으로 사용 할 경우(예: /audio 하위 파일 사용) 원본 경로에 /audio와 같이 명시할 수 있습니다.
3. 캐싱 설정: CDN에서 원본 파일의 캐싱 전략을 선택하는 메뉴입니다.(별도 선택을 하지 않고 진행하겠습니다.)
4. viewer 전송 설정: 가장 중요한 설정 화면입니다. secure token은 CDN별로 제공되는 보안 key를 사용해야합니다. Secure Token의 상태를 '사용'으로 변경합니다. Token은 query string과 cookie두 가지 방식중에 하나를 선택할 수 있습니다. 이번 예제에서는 query string 방식을 선택하겠습니다.
모든 설정을 확정한 이후 10분여를 기다리면 CDN이 생성됩니다. 생성된 CDN을 클릭하여 서비스 도메인과 보안 key가 정상적으로 발급되었는지 확인해보겠습니다.
정상적으로 생성된 모습입니다. 이제 발급된 서비스 도메인을 이용하여 원본 요청을 할 수 있습니다. bucket에 저장되어있는 sample-audio.mp3파일을 요청하려면 https://exmaqqjmiuxm27633866.gcdn.ntruss.com/sample-audio.mp3와 같이 사용하면 됩니다. 하지만 원본 파일이 공개되어있지 않아 에러 페이지가 노출됩니다.
Secure Token 발급
이제 발급된 CDN 보안 키를 사용하여 secure token을 생성해보겠습니다. 저는 Java 개발자이지만 예제에서는 python 언어를 사용하는 예제를 진행해보겠습니다.(타 언어에 비하여 생성이 쉽고 빠르게 테스트 해볼 수 있습니다. java언어를 사용한 발급 방법이 궁금하시다면 가이드문서를 참고해주세요.)
python 토큰 발급 예제 코드
import binascii
import hashlib
import hmac
import optparse
import os
import re
import sys
import time
if sys.version_info[0] >= 3:
from urllib.parse import quote_plus
else:
from urllib import quote_plus
# Force the local timezone to be GMT.
os.environ['TZ'] = 'GMT'
class EdgeAuthError(Exception):
def __init__(self, text):
self._text = text
def __str__(self):
return 'EdgeAuthError:{0}'.format(self._text)
def _getText(self):
return str(self)
text = property(_getText, None, None,
'Formatted error text.')
class EdgeAuth:
def __init__(self, token_type=None, token_name='__token__',
key=None, algorithm='sha256', salt=None,
ip=None, payload=None, session_id=None,
start_time=None, end_time=None, window_seconds=None,
field_delimiter='~', acl_delimiter='!',
escape_early=False, verbose=False):
if key is None or len(key) <= 0:
raise EdgeAuthError('You must provide a secret in order to '
'generate a new token.')
self.token_type = token_type
self.token_name = token_name
self.key = key
self.algorithm = algorithm
self.salt = salt
self.ip = ip
self.payload = payload
self.session_id = session_id
self.start_time = start_time
self.end_time = end_time
self.window_seconds = window_seconds
self.field_delimiter = field_delimiter
self.acl_delimiter = acl_delimiter
self.escape_early = escape_early
self.verbose = verbose
def _escape_early(self, text):
if self.escape_early:
def toLower(match):
return match.group(1).lower()
return re.sub(r'(%..)', toLower, quote_plus(text))
else:
return text
def _generate_token(self, path, is_url):
start_time = self.start_time
end_time = self.end_time
if str(start_time).lower() == 'now':
start_time = int(time.mktime(time.gmtime()))
elif start_time:
try:
if int(start_time) <= 0:
raise EdgeAuthError('start_time must be ( > 0 )')
except:
raise EdgeAuthError('start_time must be numeric or now')
if end_time:
try:
if int(end_time) <= 0:
raise EdgeAuthError('end_time must be ( > 0 )')
except:
raise EdgeAuthError('end_time must be numeric')
if self.window_seconds:
try:
if int(self.window_seconds) <= 0:
raise EdgeAuthError('window_seconds must be ( > 0 )')
except:
raise EdgeAuthError('window_seconds must be numeric')
if end_time is None:
if self.window_seconds:
if start_time is None:
# If we have a window_seconds without a start time,
# calculate the end time starting from the current time.
end_time = int(time.mktime(time.gmtime())) + \
self.window_seconds
else:
end_time = start_time + self.window_seconds
else:
raise EdgeAuthError('You must provide an expiration time or '
'a duration window ( > 0 )')
if start_time and (end_time <= start_time):
raise EdgeAuthError('Token will have already expired.')
if self.verbose:
print('''
Akamai Token Generation Parameters
Token Type : {0}
Token Name : {1}
Key/Secret : {2}
Algo : {3}
Salt : {4}
IP : {5}
Payload : {6}
Session ID : {7}
Start Time : {8}
End Time : {9}
Window(seconds) : {10}
Field Delimiter : {11}
ACL Delimiter : {12}
Escape Early : {13}
PATH : {14}
Generating token...'''.format(self.token_type if self.token_type else '',
self.token_name if self.token_name else '',
self.key if self.key else '',
self.algorithm if self.algorithm else '',
self.salt if self.salt else '',
self.ip if self.ip else '',
self.payload if self.payload else '',
self.session_id if self.session_id else '',
start_time if start_time else '',
end_time if end_time else '',
self.window_seconds if self.window_seconds else '',
self.field_delimiter if self.field_delimiter else '',
self.acl_delimiter if self.acl_delimiter else '',
self.escape_early if self.escape_early else '',
('url: ' if is_url else 'acl: ') + path))
hash_source = []
new_token = []
if self.ip:
new_token.append('ip={0}'.format(self._escape_early(self.ip)))
if start_time:
new_token.append('st={0}'.format(start_time))
new_token.append('exp={0}'.format(end_time))
if not is_url:
new_token.append('acl={0}'.format(path))
if self.session_id:
new_token.append('id={0}'.format(self._escape_early(self.session_id)))
if self.payload:
new_token.append('data={0}'.format(self._escape_early(self.payload)))
hash_source = list(new_token)
if is_url:
hash_source.append('url={0}'.format(self._escape_early(path)))
if self.salt:
hash_source.append('salt={0}'.format(self.salt))
if self.algorithm.lower() not in ('sha256', 'sha1', 'md5'):
raise EdgeAuthError('Unknown algorithm')
token_hmac = hmac.new(
binascii.a2b_hex(self.key.encode()),
self.field_delimiter.join(hash_source).encode(),
getattr(hashlib, self.algorithm.lower())).hexdigest()
new_token.append('hmac={0}'.format(token_hmac))
return self.field_delimiter.join(new_token)
def generate_acl_token(self, acl):
if not acl:
raise EdgeAuthError('You must provide acl')
elif isinstance(acl, list):
acl = self.acl_delimiter.join(acl)
return self._generate_token(acl, False)
def generate_url_token(self, url):
if not url:
raise EdgeAuthError('You must provide url')
return self._generate_token(url, True)
if __name__ == '__main__':
parser = optparse.OptionParser()
parser.add_option(
'-t', '--token_type',
action='store', type='string', dest='token_type',
help='Select a preset. (Not Supported Yet)')
parser.add_option(
'-n', '--token_name',
action='store', default='__token__', type='string', dest='token_name',
help='Parameter name for the new token. [Default: __token__]')
parser.add_option(
'-k', '--key',
action='store', type='string', dest='key',
help='Secret required to generate the token. It must be hexadecimal digit string with even-length.')
parser.add_option(
'-A', '--algo',
action='store', type='string', dest='algorithm', default='sha256',
help='Algorithm to use to generate the token. (sha1, sha256, or md5) [Default:sha256]')
parser.add_option(
'-S', '--salt',
action='store', type='string', dest='salt',
help='Additional data validated by the token but NOT included in the token body.')
parser.add_option(
'-s', '--start_time',
action='store', type='string', dest='start_time',
help="What is the start time? (Use 'now' for the current time)")
parser.add_option(
'-e', '--end_time',
action='store', type='string', dest='end_time',
help='When does this token expire? --end_time overrides --window')
parser.add_option(
'-w', '--window',
action='store', type='int', dest='window_seconds',
help='How long is this token valid for?')
parser.add_option(
'-d', '--field_delimiter',
action='store', default='~', type='string', dest='field_delimiter',
help='Character used to delimit token body fields. [Default:~]')
parser.add_option(
'-D', '--acl_delimiter',
action='store', default='!', type='string', dest='acl_delimiter',
help='Character used to delimit acl fields. [Default:!]')
parser.add_option(
'-x', '--escape_early',
action='store_true', default=False, dest='escape_early',
help='Causes strings to be url encoded before being used.')
parser.add_option(
'-v', '--verbose',
action='store_true', default=False, dest='verbose',
help='Print all arguments.')
parser.add_option(
'-u', '--url',
action='store', type='string', dest='url',
help='URL path. [Used for:URL]')
parser.add_option(
'-a', '--acl',
action='store', type='string', dest='access_list',
help='Access control list delimited by ! [ie. /*]')
parser.add_option(
'-i', '--ip',
action='store', type='string', dest='ip_address',
help='IP Address to restrict this token to. IP Address to restrict this token to. \
(Troublesome in many cases (roaming, NAT, etc) so not often used)')
parser.add_option(
'-p', '--payload',
action='store', type='string', dest='payload',
help='Additional text added to the calculated digest.')
parser.add_option(
'-I', '--session_id',
action='store', type='string', dest='session_id',
help='The session identifier for single use tokens or other advanced cases.')
(options, args) = parser.parse_args()
generator = EdgeAuth(
token_type=options.token_type,
token_name=options.token_name,
key=options.key,
algorithm=options.algorithm,
salt=options.salt,
ip=options.ip_address,
payload=options.payload,
session_id=options.session_id,
start_time=options.start_time,
end_time=options.end_time,
window_seconds=options.window_seconds,
field_delimiter=options.field_delimiter,
acl_delimiter=options.acl_delimiter,
escape_early=options.escape_early,
verbose=options.verbose)
url=options.url
acl=options.access_list
if (url and acl):
print("You should input one in the 'url' or the 'acl'.")
else:
if acl:
token = generator.generate_acl_token(acl)
else: # url
token = generator.generate_url_token(url)
print("### Cookie or Query String ###")
print("{0}={1}".format(options.token_name, token))
print("### Header ###")
print("{0}: {1}".format(options.token_name, token))네이버에서 제공하는 python 토큰 생성 코드입니다. .py파일에 해당 내용을 붙여넣어주세요.
이후 터미널로 아래와같이 명령어를 입력하면 토큰이 생성됩니다.
$ python cms_edgeauth.py -k b2b1 -n token -s now -w 3600 -a /sample.pdf* 예시와 같이 입력어 입력
=> 다음과 같은 결과 출력
token=st=1592204787~exp=1592208387~acl=/sample.pdf*~hmac=79872098f16596c8c40ebab649ae2aac8cce3e3bece204b641c99b6cfac42779- 옵션
- -k: CDN에서 발급 받은 key값
- -n: (고정값)token
- -s: 토큰 유효 시작 시간을 명시합니다.
- -w: 토큰 활성화 시간(초)
- -a: 권한을 부여할 파일/디렉토리를 명시합니다.
이제 실제 secure token을 발급하여 앞전에 업로드하였던 sample-audio.mp3 파일을 조회해보겠습니다.
$ python test.py -k cc71dee9b21199c14784 -n token -s now -w 3600 -a '/sample-audio.mp3*'
### Cookie or Query String ###
token=st=1738757016~exp=1738760616~acl=/sample-audio.mp3*~hmac=f4916ab9109e25d3b62ba6e783aa99288f9e43dd432600e81c273bd82ba090d4
### Header ###
token: st=1738757016~exp=1738760616~acl=/sample-audio.mp3*~hmac=f4916ab9109e25d3b62ba6e783aa99288f9e43dd432600e81c273bd82ba090d4서비스 도메인을 사용하여 /sample-audio.mp3파일을 요청하고 발급받은 토큰값을 query string으로 전달해보겠습니다.
정상적으로 음원 파일이 조회되었습니다.
마무리
NCP는 AWS S3 SDK를 사용할 수 있기 때문에 이 방법 외에도 Presigned URL방식도 고려해 볼 수 있습니다. 만약 object storage의 공개 여부를 공개로 설정해야만 하는 제약이 있다면 버킷으로 파일 업로드 시 파일 암호화를 통하여 관리하는 방법도 고려해볼 수 있겠습니다. 자세한 사용 방법은 가이드문서를 참고해주시면 감사드리겠습니다.
