현재 진행하고 있는 프로젝트에서 앱에 게시된 보상형 광고를 사용자가 봤을 경우, 이에 대한 보상을 주기 위해 구글에서 SSV 콜백을 서버로 보내주면 이를 확인하는 API를 구현해야했다. 구글에서 제공하는 문서에서는 이를 Java로 구현하였고, Tink라는 라이브러리를 사용하고 있었다.
따라서, 위 링크에 나와있는 SSV 수동 인증 로직을 차례대로 Python으로 구현해보았다. 일단 임포트해야 할 모듈 목록이다.
import json
import base64
import urllib.parse
import urllib.request
import hashlib
from ecdsa.keys import VerifyingKey, BadSignatureError
from ecdsa.util import sigdecode_der공개 키 가져오기
SSV 콜백을 확인하는데 사용하는 공개 키 목록은 구글에서 제공하고 있다. 아래 URL을 통해 JSON 형식으로 제공하는데, 이를 파싱하여 딕셔너리 타입으로 반환하는 함수를 구현하자.
https://gstatic.com/admob/reward/verifier-keys.json
{
"keys": [
{
keyId: 1916455855,
pem: "-----BEGIN PUBLIC KEY-----\nMF...YTPcw==\n-----END PUBLIC KEY-----"
base64: "MFkwEwYHKoZIzj0CAQYI...ltS4nzc9yjmhgVQOlmSS6unqvN9t8sqajRTPcw=="
},
{
keyId: 3901585526,
pem: "-----BEGIN PUBLIC KEY-----\nMF...aDUsw==\n-----END PUBLIC KEY-----"
base64: "MFYwEAYHKoZIzj0CAQYF...4akdWbWDCUrMMGIV27/3/e7UuKSEonjGvaDUsw=="
},
],
}ADMOB_KEY_SERVER_URL = 'https://www.gstatic.com/admob/reward/verifier-keys.json'
def parse_public_keys(): # 구글 Admob key server 에서 공개키 목록 가져오기
keys = dict()
response = urllib.request.urlopen(ADMOB_KEY_SERVER_URL)
if response.status != 200:
return keys
keys_json = response.read().decode('utf-8')
if not keys_json:
return keys
keys_data = json.loads(keys_json)
if not keys_data or 'keys' not in keys_data:
return keys
for key_ids in keys_data['keys']: # 키 목록에서 keyId를 키로 하는 딕셔너리 생성
keys[str(key_ids['keyId'])] = dict(
pem=key_ids['pem'],
base64=key_ids['base64']
)
return keys반환되는 keys 변수에는 딕셔너리 타입으로 공개 키 목록이 저장되어 있다.
인증할 내용과 signature 및 key_id 가져오기
구글에서 보내는 SSV 콜백 URL은 아래와 같은 구조로 되어 있는데, 진하게 되어 있는 부분이 바로 인증할 내용이고, 마지막 두 개의 파라미터가 각각 signature 와 key_id 이다.
URL 파라미터 중에서 인증할 내용을 파싱하고, signature 와 key_id 를 저장하는 함수를 구현해 보자.
def parse_query_string(query_string): # SSV 콜백 URL의 쿼리 스트링에서 인증할 내용, signature 및 key id 값 가져오기
query = dict(
signature='',
key_id='',
message=''
)
if not query_string or query_string.find('&') == -1:
return query
query_string_dict = dict([x.split('=', 1) for x in query_string.split('&')])
query['signature'] = query_string_dict.get('signature').strip() or ''
query['message'] = query_string[:query_string.index('signature') - 1].strip() or ''
query['key_id'] = query_string_dict.get('key_id', '').strip() or ''
return query해당 함수를 호출할 때 query_string 값에는 SSV 콜백 URL에서 파라미터 부분만 추출한 문자열 값이 전달되도록 해야 한다.
인증 하기
def signature_verifier(ver_message, ver_signature, public_key_pem):
public_key = VerifyingKey.from_pem(public_key_pem)
ver_signature = base64.urlsafe_b64decode(ver_signature + '=' * (4 - len(ver_signature) % 4))
ver_message = bytes(ver_message, encoding="utf8")
try:
return public_key.verify(
ver_signature,
ver_message,
hashfunc=hashlib.sha256,
sigdecode=sigdecode_der,
)
except BadSignatureError as e:
return False공개 키 목록에서 pem 값으로 확인키를 생성한다. 생성한 확인키를 가지고 signature 값이 message 를 서명한 값인지 확인하는 과정을 거쳐야 하는데, 이 signature 값은 URL 파라미터 값으로 전달되므로 URL safe한 base64로 인코딩된 값이다. 따라서 이를 urlsafe_b64decode 메소드로 디코딩한 후에 확인하는 과정을 거쳐야 한다. 암호화에 대해서 미숙하기 때문에 ecdsa 모듈 사용법은 구글링을 통해 알아낸 거라 설명할 수 있는게 많이 없어서 일단 저런 로직으로 구현된다는 것만 알아두자.
메인 로직에서는 요청 URL을 통해 쿼리 스트링을 파싱하여 이를 parse_query_string() 함수를 호출할 때 인자 값으로 전달한다. 반환 후 인증할 내용, signature, key_id 값을 구하고, key_id 값에 해당하는 공개 키를 가져와 확인하는 함수를 호출한다.
query_string = request.query_string.decode('utf-8')
query_dict = parse_query_string(query_string)
signature = query_dict.get('signature').strip()
message = query_dict.get('message').strip()
key_id = query_dict.get('key_id').strip()
verifier_keys = parse_public_keys()
pem_keys = verifier_keys.get(str(key_id), None)
verify_result = signature_verifier(message, signature, pem_keys['pem'])