두둥. 모바일 팀에서 전달해준 메시지.
FCM HTTP API 의 지원이 2023년 6월 20일부터 중단되고, 2024년 6월 1일부터 기존 API 로 모바일 푸시 알림 전송이 불가해짐. HTTP v1 으로 마이그레이션을 해야했음
기존 HTTP에서 HTTP v1로 마이그레이션 | Firebase 클라우드 메시징
FCM API 를 직접 개발하진 않았지만 관련 커밋에 내 이름이 있다는 이유로 나에게 넘어온 전환 미션.
해야지 뭐.
🕹️ v1 으로 바뀌면서 변경되는 점
docs said..
- 액세스 토큰을 통한 보안 향상: HTTP v1 API는 OAuth2 보안 모델에 따라 수명이 짧은 액세스 토큰을 사용합니다. 액세스 토큰이 공개되는 경우에도 악의적으로 사용될 수 있는 시간은 만료되기 전 1시간 정도뿐입니다. 갱신 토큰이 기존 API에서 사용하는 보안 키만큼 자주 전송되지 않으므로 캡처될 가능성이 매우 낮습니다.
- 보다 효율적인 플랫폼에 따른 메시지 맞춤설정: 메시지 본문의 경우 HTTP v1 API는 모든 대상 인스턴스에 전달되는 공용 키는 물론 플랫폼에 따라 메시지를 맞춤설정할 수 있는 플랫폼별 키가 있습니다. 이러한 키를 사용하면 메시지 하나로 여러 클라이언트 플랫폼에 약간 다른 페이로드를 전송하는 '재정의'를 만들 수 있습니다.
- 새 클라이언트 플랫폼 버전을 위한 확장성 강화 및 미래 경쟁력 확보: HTTP v1 API는 Apple 플랫폼, Android, 웹에 제공되는 메시지 옵션을 완전히 지원합니다. 플랫폼별로 JSON 페이로드에 자체 정의된 블록이 있으므로 FCM에서 필요에 따라 새 버전과 새 플랫폼으로 API를 확장할 수 있습니다.
엔드포인트
엔드포인트 자체도 바뀌었고, project_id 를 같이 담아줘야함
# 기존
POST https://fcm.googleapis.com/fcm/send
# 변경
POST https://fcm.googleapis.com/v1/projects/{{project_id}}/messages:send인증방식
기존에는 서버 토큰을 하나 받아놓고 해당 토큰을 Authorization 으로 넘겨서 인증을 받는 방식이었으나, Bearer <valid Oauth 2.0 token> 으로 변경됨
이에 따라 요청 시 마다 google auth 를 통해 인증 토큰을 발급 받아야 함
# 기존
Authorization: key=AIzaSyZ-1u...0GBYzPu7Udno5aA
# 변경
Authorization: Bearer ya29.ElqKBGN2Ri_Uz...HnS_uNreA그룹 메시징 없어짐
기존에는 registration_ids 라는 파라미터에 발송 대상 푸시 토큰 키들을 리스트로 담으면 한 번에 발송하는 기능을 지원했으나, v1 에서는 지원하지 않음.
Firebase Admin SDK 를 사용하면 발송할 수 있다고 하나, 현재 서비스 기능 상 그룹 메시징 기능은 필요하지 않아 v1 API를 활용해 발송하기로 결정
발송 바디 형식
## 특정 푸시 토큰(유저) 대상 발송 시
data = {
'link': 'xxxx',
}
notification = {
'title': app,
'body': msg,
}
# 기존
{
'registration_ids': push_keys, # []
'notification': notification,
'data': data,
}
# 변경
{
'message': {
'token': push_key,
'notification': notification,
'data': data,
}
}🐶🦶개발
참조
헬퍼화 및 클래스 초기화
여러 레포 및 시스템에서 사용될 것을 고려하여 공통 헬퍼로 작업하고, 클래스 초기화를 이용해 어떤 앱에서 사용하는지 지정.
추후 앱 추가 시 enum과 const 만 추가하면 자동 대응 되도록 작성
# enum.py
class AppType(str, Enum):
""" 앱 타입
TEST: 테스트
"""
TEST: str = constants.APP_TYPE.get('TEST')
class FcmProjectIdType(str, Enum):
""" FCM project id type
secretmanager 에 저장된 키 기준으로 키 설정
test: 테스트
"""
test: str = constants.FCM_PROJECT_ID_TYPE.get('TEST')
class FcmSecretManagerKeyType(str, Enum):
""" FCM SECRET MANAGER KEY type
secretmanager 에서 불러오는 객체 타입
secretmanager 에 저장된 키 기준으로 키 설정
test: 테스트
"""
test: str = constants.FCM_SECRET_MANAGER_KEY_TYPE.get('TEST')
# fcm.py
class FCMHelper:
""" GOOGLE FCM 헬퍼
"""
def __init__(self, app_type: AppType):
""" 헬퍼 초기화
추후 다른 프로젝트에서 사용할 경우를 대비해 app_type 받아서 초기화 함
"""
self.app_type = app_type
if app_type == AppType.TEST:
self.project_id = FcmProjectIdType[app_type.value].value
self.fcm_secretmanager_key = FcmSecretManagerKeyType[app_type.value].value
else:
raise Exception('유효하지 않은 앱이 설정되었습니다.')
self.base_url = 'https://fcm.googleapis.com'
self.fcm_endpoint = 'v1/projects/' + self.project_id + '/messages:send'
self.fcm_url = self.base_url + '/' + self.fcm_endpoint
self.scopes = ['https://www.googleapis.com/auth/firebase.messaging']FCM 인증 요청
서버가 구글 이외의 서버 환경에서 실행되기 때문에 Firebase 프로젝트에서 서비스 계정 JSON 파일을 받아 인증에 사용해야함. 이 정보를 이용해 구글에 액세스 토큰을 요청하고 액세스 토큰이 만료되면 토큰 갱신 메서드가 자동으로 호출되어 업데이트된 액세스 토큰이 발급됩니다.
- json 파일 다운로드 방법 참조
- json 양식
{ "type": "service_account", "project_id": "sparkplus-app", "private_key_id": "xxxxxxx", "private_key": "-----BEGIN PRIVATE KEY-----\nxxxxxxo\n-----END PRIVATE KEY-----\n", "client_email": "xxxxx.gserviceaccount.com", "client_id": "1111111", "auth_uri": "https://accounts.google.com/o/oauth2/auth", "token_uri": "https://oauth2.googleapis.com/token", "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs", "client_x509_cert_url": "https://www.googleapis.com/robot/v1/xxxxx.gserviceaccount.com", "universe_domain": "googleapis.com" }
문서에서는 json 파일 위치를 환경변수로 지정하고 읽어서 사용하는 방법을 제시했는데, 여기서는 기존에 암호화된 내용을 관리하는 방식인 SecretManager 에 json 내용을 저장해서 불러와서 사용하도록 했음
- 문서에서 제시했던 방식 (
from_service_account_file사용) GOOGLE_APPLICATION_CREDENTIALS 환경 변수를 서비스 계정 키가 포함된 JSON 파일의 파일 경로로 설정합니다. 이 변수는 현재 셸 세션에만 적용되므로 새 세션을 열면 변수를 다시 설정합니다.
위 단계를 완료하면 애플리케이션 기본 사용자 인증 정보(ADC)가 암묵적으로 사용자 인증 정보를 확인할 수 있으므로 Google 이외의 환경에서 테스트 또는 실행할 때 서비스 계정의 사용자 인증 정보를 사용할 수 있습니다.# linux or macOS export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json" # window $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\service-account-file.json"# 파일을 읽어서 credentials 틀 만들어줌 credentials = service_account.Credentials.from_service_account_file( 'service-account.json', scopes=SCOPES)
파일을 읽는 메서드 말고 dictionary 에서 내용을 읽는 메서드가 제공되고 있어서 해당 메서드로 credentials 가공(from_service_account_info)
async def _get_access_token(self):
"""Retrieve a valid access token that can be used to authorize requests.
:return: Access token.
"""
# ----------------------------------------------------------------------------------------
# 셋팅
# ----------------------------------------------------------------------------------------
app = self.app_type
get_fcm = get_secret(key=self.fcm_secretmanager_key, is_once=True)
if app not in get_fcm:
raise ApiException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, message='설정된 앱키가 없습니다.')
fcm_key = get_fcm.get(app)
try:
credentials = service_account.Credentials.from_service_account_info(
fcm_key, scopes=self.scopes)
request = google.auth.transport.requests.Request()
credentials.refresh(request)
return credentials.token
except Exception as e:
print(e)비동기 고려
dependency 로 google-auth==2.29.0 를 추가하고 빌드를 했으나 requests import error 가 발생함. 기존에는 http 요청을 비동기 모듈인 aiohttp 로 진행하고 있어서 requests 가 존재하지 않았음
requests 는 비동기를 지원하지 않아서 google auth 가 어떤 방식으로 인증을 요청하는지 확인해봄
google auth 에서는 request session 을 google/auth/transport/requests 에서 만들어주고 있는데, 여기서 세션을 만들 때 requests 를 이용함
# requests/Request/__init__
def __init__(self, session=None):
if not session:
session = requests.Session()
self.session = session
# requests/Request/__call__
response = self.session.request( # requests/session/request
method, url, data=body, headers=headers, timeout=timeout, **kwargs
)라이브러리의 비동기 지원여부를 확인해보니 google/auth/transport/_aiohttp_requests 를 통해 지원하긴 함.
내부적으로 aiohttp 를 통해 세션을 만들어서 요청함
# _aiohttp_requests/Request/__call_
if self.session is None: # pragma: NO COVER
self.session = aiohttp.ClientSession(
auto_decompress=False
) # pragma: NO COVER
requests._LOGGER.debug("Making request: %s %s", method, url)
response = await self.session.request(
method, url, data=body, headers=headers, timeout=timeout, **kwargs
)비동기로 작성을 할 수는 있으나 일단 비동기 지원 자체가 실험적인 기능이고, 바뀔 수 있다고 함
# _aiottp_requests
"""Transport adapter for Async HTTP (aiohttp).
NOTE: This async support is experimental and marked internal. This surface may
change in minor releases.
"""그리고 session 이 with 문 등으로 감싸져있지 않아서 자동으로 닫히지 않아 세션 누수 이슈도 있어서 기본으로 지원하는 동기 방식으로 사용하기로 결정
-끝-
