ABDM changes for consuming APIs in the mobile app.

corehq/apps/domain/tests/test_deletion_models.py

@@ -39,6 +39,7 @@
     'toggle_ui',
     'sso',
     'oauth_integrations',
+    'rest_framework',
 }
 
 IGNORE_MODELS = {
@@ -65,6 +66,7 @@
     'util.TransientBounceEmail',
     'registration.AsyncSignupRequest',
     'users.UserHistory',
+    'custom.abdm.models.ABDMUser',
 }
 
 

corehq/apps/dump_reload/tests/test_dump_models.py

@@ -100,6 +100,7 @@
     "util.PermanentBounceMeta",
     "util.TransientBounceEmail",
     "users.Permission",
+    "abdm.ABDMUser",
 }
 
 # TODO: determine which of these should not be ignored

corehq/ex-submodules/casexml/apps/phone/xml.py

@@ -8,6 +8,7 @@
 import six
 
 from casexml.apps.phone.exceptions import RestoreException
+from custom.abdm.milestone_one.utils.user_util import get_abdm_api_token
 
 USER_REGISTRATION_XMLNS_DEPRECATED = "http://openrosa.org/user-registration"
 USER_REGISTRATION_XMLNS = "http://openrosa.org/user/registration"
@@ -129,10 +130,16 @@ def get_registration_element_data(restore_user):
         "password": restore_user.password,
         "uuid": restore_user.user_id,
         "date": date_to_xml_string(restore_user.date_joined),
-        "user_data": restore_user.user_session_data
+        "user_data": get_session_data_with_abdm_token(restore_user)
     }
 
 
+def get_session_data_with_abdm_token(restore_user):
+    user_data = {"abdm_api_token": get_abdm_api_token(restore_user.username)}
+    user_data.update(restore_user.user_session_data)
+    return user_data
+
+
 # Case registration blocks do not have a password
 def get_registration_element_for_case(case):
     root = safe_element("Registration")

custom/abdm/auth.py

@@ -0,0 +1,19 @@
+from rest_framework.authentication import TokenAuthentication
+from rest_framework.exceptions import AuthenticationFailed
+from custom.abdm.models import ABDMUser
+
+
+class ABDMUserAuthentication(TokenAuthentication):
+
+    def authenticate(self, request):
+        access_token = request.META.get('HTTP_AUTHORIZATION', "")
+        token = access_token.split(" ")[-1]
+        if not token:
+            return None
+        try:
+            user = ABDMUser.objects.get(access_token=token)
+        except ABDMUser.DoesNotExist:
+            raise AuthenticationFailed('Unauthorized')
+        if not user.is_token_valid:
+            raise AuthenticationFailed('Invalid Token')
+        return (user, None)

custom/abdm/migrations/0001_squashed_0002_auto_20220926_0547.py

@@ -0,0 +1,23 @@
+# Generated by Django 3.2.15 on 2022-09-26 05:56
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    replaces = [('abdm', '0001_initial'), ('abdm', '0002_auto_20220926_0547')]
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ABDMUser',
+            fields=[
+                ('username', models.CharField(max_length=100, primary_key=True, serialize=False)),
+                ('access_token', models.CharField(blank=True, max_length=2000, null=True)),
+            ],
+        ),
+    ]

custom/abdm/milestone_one/utils/abha_creation_util.py

@@ -0,0 +1,44 @@
+import logging
+
+from custom.abdm.milestone_one.utils.request_util import get_response_http_post
+
+logger = logging.getLogger(__name__)
+
+
+def generate_aadhar_otp(aadhaar_number):
+    generate_aadhar_otp_url = "v1/registration/aadhaar/generateOtp"
+    payload = {"aadhaar": str(aadhaar_number)}
+    return get_response_http_post(generate_aadhar_otp_url, payload)
+
+
+def generate_mobile_otp(mobile_number, txnid):
+    generate_mobile_otp_url = "v1/registration/aadhaar/generateMobileOTP"
+    payload = {"mobile": str(mobile_number), "txnId": txnid}
+    return get_response_http_post(generate_mobile_otp_url, payload)
+
+
+def verify_aadhar_otp(otp, txnid):
+    verify_aadhaar_otp_url = "v1/registration/aadhaar/verifyOTP"
+    payload = {"otp": str(otp), "txnId": txnid}
+    return get_response_http_post(verify_aadhaar_otp_url, payload)
+
+
+def verify_mobile_otp(otp, txnid):
+    verify_mobile_otp_url = "v1/registration/aadhaar/verifyMobileOTP"
+    payload = {"otp": str(otp), "txnId": txnid}
+    return get_response_http_post(verify_mobile_otp_url, payload)
+
+
+def create_health_id(txnid):
+    create_health_id_url = "v1/registration/aadhaar/createHealthIdWithPreVerified"
+    payload = {
+        "email": "",
+        "firstName": "",
+        "healthId": "",
+        "lastName": "",
+        "middleName": "",
+        "password": "",
+        "profilePhoto": "",
+        "txnId": txnid
+    }
+    return get_response_http_post(create_health_id_url, payload)

custom/abdm/milestone_one/utils/abha_verification_util.py

@@ -0,0 +1,36 @@
+import requests
+
+from custom.abdm.milestone_one.utils.request_util import get_access_token, get_response_http_post, base_url
+
+
+def generate_auth_otp(health_id, auth_method):
+    auth_otp_url = "v2/auth/init"
+    payload = {"authMethod": auth_method, "healthid": health_id}
+    return get_response_http_post(auth_otp_url, payload)
+
+
+def confirm_with_mobile_otp(otp, txn_id):
+    confirm_with_mobile_otp_url = "v1/auth/confirmWithMobileOTP"
+    payload = {"otp": otp, "txnId": txn_id}
+    return get_response_http_post(confirm_with_mobile_otp_url, payload)
+
+
+def confirm_with_aadhaar_otp(otp, txn_id):
+    confirm_with_aadhaar_otp_url = "v1/auth/confirmWithAadhaarOtp"
+    payload = {"otp": otp, "txnId": txn_id}
+    return get_response_http_post(confirm_with_aadhaar_otp_url, payload)
+
+
+def get_account_information(x_token):
+    account_information_url = "v1/account/profile"
+    headers = {"Content-Type": "application/json; charset=UTF-8"}
+    token = get_access_token()
+    headers.update({"Authorization": "Bearer {}".format(token), "X-Token": f"Bearer {x_token}"})
+    resp = requests.get(url=base_url + account_information_url, headers=headers)
+    return resp.json()
+
+
+def search_by_health_id(health_id):
+    search_by_health_id_url = "v1/search/searchByHealthId"
+    payload = {"healthId": health_id}
+    return get_response_http_post(search_by_health_id_url, payload)

custom/abdm/milestone_one/utils/decorators.py

@@ -0,0 +1,37 @@
+import logging
+
+from functools import wraps
+from typing import List
+
+from custom.abdm.milestone_one.utils.response_handler import generate_invalid_req_response
+
+logger = logging.getLogger(__name__)
+
+
+def required_request_params(params):
+    """
+    Checks if the parameters provided in the decorator(a list of strings) are present in the DRF request.
+    If not, raises 400 Bad Request error.
+    """
+
+    def decorate(fn):
+        if not (params and isinstance(params, List)):
+            error_msg = "Request could not be validated as a valid input not provided. \
+                Required: List of parameters."
+            logger.warning(error_msg)
+            return generate_invalid_req_response(error_msg)
+
+        @wraps(fn)
+        def wrapped(request, *args, **kwargs):
+            invalid_params = []
+            for param in params:
+                if not request.data.get(param):
+                    invalid_params.append(param)
+            if invalid_params:
+                error_msg = f"Missing required parameter(s) in the request: {','.join(invalid_params)}"
+                return generate_invalid_req_response(error_msg)
+            return fn(request, *args, **kwargs)
+
+        return wrapped
+
+    return decorate

custom/abdm/milestone_one/utils/request_util.py

@@ -0,0 +1,28 @@
+import json
+from typing import Any, Dict
+import requests
+
+from django.conf import settings
+
+base_url = "https://healthidsbx.abdm.gov.in/api/"
+gateway_url = "https://dev.abdm.gov.in/gateway/v0.5/sessions"
+
+
+def get_access_token():
+    payload = {"clientId": settings.ABDM_CLIENT_ID, "clientSecret": settings.ABDM_CLIENT_SECRET}
+    headers = {"Content-Type": "application/json; charset=UTF-8"}
+    resp = requests.post(url=gateway_url, data=json.dumps(payload), headers=headers)
+    if resp.status_code == 200:
+        return resp.json().get("accessToken")
+
+
+def get_response_http_post(api_url, payload, additional_headers):
+    req_url = base_url + api_url
+    headers = {"Content-Type": "application/json"}
+    token = get_access_token()
+    headers.update({"Authorization": "Bearer {}".format(token)})
+    if additional_headers:
+        headers.update(additional_headers)
+    resp = requests.post(url=req_url, data=json.dumps(payload), headers=headers)
+    resp.raise_for_status()
+    return resp.json()

custom/abdm/milestone_one/utils/response_handler.py

@@ -0,0 +1,40 @@
+from rest_framework.response import Response
+
+from rest_framework.status import HTTP_200_OK, HTTP_500_INTERNAL_SERVER_ERROR, HTTP_400_BAD_REQUEST
+
+success_response_keys = ["txnId", "healthIdNumber", "auth_methods", "token"]
+
+
+def get_response(response_data):
+    if not response_data:
+        return Response({"error": "No valid response found"}, status=HTTP_500_INTERNAL_SERVER_ERROR)
+    if any([key in response_data for key in success_response_keys]):
+        return _get_success_abdm_response(response_data)
+    return _get_error_abdm_response(response_data)
+
+
+def generate_invalid_req_response(message):
+    resp = {
+        "code": "400",
+        "message": "Unable to process the current request due to incorrect data entered.",
+        "details": [{
+            "message": message,
+            "attribute": None
+        }]
+    }
+    return Response(resp, status=HTTP_400_BAD_REQUEST)
+
+
+def _get_success_abdm_response(response_data):
+    return Response(response_data, status=HTTP_200_OK)
+
+
+def _get_error_abdm_response(response_data):
+    if "code" in response_data:
+        return _parse_response(response_data)
+    return generate_invalid_req_response(response_data)
+
+
+def _parse_response(response_data):
+    status_code = int(response_data.get("code").split("-")[-1])
+    return Response(response_data, status=status_code)

custom/abdm/milestone_one/utils/user_util.py

@@ -0,0 +1,12 @@
+import logging
+
+from custom.abdm.models import ABDMUser
+
+logger = logging.getLogger(__name__)
+
+
+def get_abdm_api_token(username):
+    user, _ = ABDMUser.objects.get_or_create(username=username)
+    if not user.access_token:
+        user.generate_token()
+    return user.access_token

custom/abdm/milestone_one/views/abha_creation_views.py

@@ -0,0 +1,54 @@
+from rest_framework.decorators import api_view, permission_classes, authentication_classes
+from rest_framework.permissions import IsAuthenticated
+from custom.abdm.auth import ABDMUserAuthentication
+
+from custom.abdm.milestone_one.utils import abha_creation_util as abdm_util
+from custom.abdm.milestone_one.utils.decorators import required_request_params
+from custom.abdm.milestone_one.utils.response_handler import get_response
+
+
+@api_view(["POST"])
+@permission_classes((IsAuthenticated,))
+@authentication_classes((ABDMUserAuthentication,))
+@required_request_params(["aadhaar"])
+def generate_aadhaar_otp(request):
+    aadhaar_number = request.data.get("aadhaar")
+    resp = abdm_util.generate_aadhar_otp(aadhaar_number)
+    return get_response(resp)
+
+
+@api_view(["POST"])
+@permission_classes((IsAuthenticated,))
+@authentication_classes((ABDMUserAuthentication,))
+@required_request_params(["txn_id", "mobile_number"])
+def generate_mobile_otp(request):
+    txn_id = request.data.get("txn_id")
+    mobile_number = request.data.get("mobile_number")
+    resp = abdm_util.generate_mobile_otp(mobile_number, txn_id)
+    return get_response(resp)
+
+
+@api_view(["POST"])
+@permission_classes((IsAuthenticated,))
+@authentication_classes((ABDMUserAuthentication,))
+@required_request_params(["txn_id", "otp"])
+def verify_aadhaar_otp(request):
+    txn_id = request.data.get("txn_id")
+    otp = request.data.get("otp")
+    resp = abdm_util.verify_aadhar_otp(otp, txn_id)
+    return get_response(resp)
+
+
+@api_view(["POST"])
+@permission_classes((IsAuthenticated,))
+@authentication_classes((ABDMUserAuthentication,))
+@required_request_params(["txn_id", "otp"])
+def verify_mobile_otp(request):
+    txn_id = request.data.get("txn_id")
+    otp = request.data.get("otp")
+    resp = abdm_util.verify_mobile_otp(otp, txn_id)
+    if resp and "txnId" in resp:
+        resp = abdm_util.create_health_id(txn_id)
+        resp.pop("token")
+        resp.pop("refreshToken")
+    return get_response(resp)