feat: Add authentication using certificates (Same flow as the mobile app)

Author: andreroggeri

.gitignore

@@ -106,3 +106,6 @@ ENV/
 
 # OSX Files
 .DS_Store
+
+*.p12
+*.json

pynubank/__init__.py

@@ -1 +1,2 @@
-from .nubank import Nubank, NuException
+from .exception import NuRequestException, NuException
+from .nubank import Nubank

pynubank/cli.py

@@ -0,0 +1,60 @@
+import os
+import random
+import string
+from getpass import getpass
+
+from colorama import init, Fore, Style
+
+from pynubank import NuException
+from pynubank.utils.certificate_generator import CertificateGenerator
+
+
+def generate_random_id() -> str:
+    return ''.join(random.choices(string.ascii_lowercase + string.digits, k=12))
+
+
+def log(message, color=Fore.BLUE):
+    print(f'{color}{Style.DIM}[*] {Style.NORMAL}{Fore.LIGHTBLUE_EX}{message}')
+
+
+def save_cert(cert, name):
+    path = os.path.join(os.getcwd(), name)
+    with open(path, 'wb') as cert_file:
+        cert_file.write(cert.export())
+
+
+def main():
+    init()
+
+    log(f'Starting {Fore.MAGENTA}{Style.DIM}PyNubank{Style.NORMAL}{Fore.LIGHTBLUE_EX} context creation.')
+
+    device_id = generate_random_id()
+
+    log(f'Generated random id: {device_id}')
+
+    cpf = input(f'[>] Enter your CPF(Numbers only): ')
+    password = getpass('[>] Enter your password (Used on the app/website): ')
+
+    generator = CertificateGenerator(cpf, password, device_id)
+
+    log('Requesting e-mail code')
+    try:
+        email = generator.request_code()
+    except NuException:
+        log(f'{Fore.RED}Failed to request code. Check your credentials!', Fore.RED)
+        exit(1)
+
+    log(f'Email sent to {Fore.LIGHTBLACK_EX}{email}{Fore.LIGHTBLUE_EX}')
+    code = input('[>] Type the code received by email: ')
+
+    cert1, cert2 = generator.exchange_certs(code)
+
+    save_cert(cert1, 'cert.p12')
+    save_cert(cert2, 'cert_crypto.p12')
+
+    print(f'{Fore.GREEN}Certificates generated successfully. (cert.pem and cert_crypto.pem)')
+    print(f'{Fore.YELLOW}Warning, keep these certificates safe (Do not share or version in git)')
+
+
+if __name__ == '__main__':
+    main()

pynubank/exception.py

@@ -0,0 +1,15 @@
+from requests import Response
+
+
+class NuException(Exception):
+
+    def __init__(self, message):
+        super().__init__(message)
+
+
+class NuRequestException(NuException):
+    def __init__(self, response: Response):
+        super().__init__(f'The request made failed with HTTP status code {response.status_code}')
+        self.url = response.url
+        self.status_code = response.status_code
+        self.response = response

pynubank/nubank.py

@@ -1,11 +1,11 @@
-import json
 import os
 import uuid
 from typing import Tuple
 
-import requests
 from qrcode import QRCode
-from requests import Response
+
+from pynubank.utils.discovery import Discovery
+from pynubank.utils.http import HttpClient
 
 PAYMENT_EVENT_TYPES = (
     'TransferOutEvent',
@@ -17,32 +17,14 @@
 )
 
 
-class NuException(Exception):
-    def __init__(self, status_code, response, url):
-        super().__init__(f'The request made failed with HTTP status code {status_code}')
-        self.url = url
-        self.status_code = status_code
-        self.response = response
-
-
 class Nubank:
-    DISCOVERY_URL = 'https://prod-s0-webapp-proxy.nubank.com.br/api/discovery'
-    DISCOVERY_APP_URL = 'https://prod-s0-webapp-proxy.nubank.com.br/api/app/discovery'
-    auth_url = None
     feed_url = None
-    proxy_list_url = None
-    proxy_list_app_url = None
     query_url = None
     bills_url = None
 
     def __init__(self):
-        self.headers = {
-            'Content-Type': 'application/json',
-            'X-Correlation-Id': 'WEB-APP.pewW9',
-            'User-Agent': 'pynubank Client - https://github.com/andreroggeri/pynubank',
-        }
-        self._update_proxy_urls()
-        self.auth_url = self.proxy_list_url['login']
+        self.client = HttpClient()
+        self.discovery = Discovery(self.client)
 
     @staticmethod
     def _get_query(query_name):
@@ -52,19 +34,11 @@ def _get_query(query_name):
         with open(path) as gql:
             return gql.read()
 
-    def _update_proxy_urls(self):
-        request = requests.get(self.DISCOVERY_URL, headers=self.headers)
-        self.proxy_list_url = json.loads(request.content.decode('utf-8'))
-        request = requests.get(self.DISCOVERY_APP_URL, headers=self.headers)
-        self.proxy_list_app_url = json.loads(request.content.decode('utf-8'))
-
     def _make_graphql_request(self, graphql_object):
         body = {
             'query': self._get_query(graphql_object)
         }
-        response = requests.post(self.query_url, json=body, headers=self.headers)
-
-        return self._handle_response(response)
+        return self.client.post(self.query_url, json=body)
 
     def _password_auth(self, cpf: str, password: str):
         payload = {
@@ -74,15 +48,13 @@ def _password_auth(self, cpf: str, password: str):
             "client_id": "other.conta",
             "client_secret": "yQPeLzoHuJzlMMSAjC-LgNUJdUecx8XO"
         }
-        response = requests.post(self.auth_url, json=payload, headers=self.headers)
-        data = self._handle_response(response)
-        return data
-
-    def _handle_response(self, response: Response) -> dict:
-        if response.status_code != 200:
-            raise NuException(response.status_code, response.json(), response.url)
+        return self.client.post(self.discovery.get_url('login'), json=payload)
 
-        return response.json()
+    def _save_auth_data(self, auth_data: dict) -> None:
+        self.client.set_header('Authorization', f'Bearer {auth_data["access_token"]}')
+        self.feed_url = auth_data['_links']['events']['href']
+        self.query_url = auth_data['_links']['ghostflame']['href']
+        self.bills_url = auth_data['_links']['bills_summary']['href']
 
     def get_qr_code(self) -> Tuple[str, QRCode]:
         content = str(uuid.uuid4())
@@ -92,36 +64,62 @@ def get_qr_code(self) -> Tuple[str, QRCode]:
 
     def authenticate_with_qr_code(self, cpf: str, password, uuid: str):
         auth_data = self._password_auth(cpf, password)
-        self.headers['Authorization'] = f'Bearer {auth_data["access_token"]}'
+        self.client.set_header('Authorization', f'Bearer {auth_data["access_token"]}')
 
         payload = {
             'qr_code_id': uuid,
             'type': 'login-webapp'
         }
 
-        response = requests.post(self.proxy_list_app_url['lift'], json=payload, headers=self.headers)
+        response = self.client.post(self.discovery.get_app_url('lift'), json=payload)
 
-        auth_data = self._handle_response(response)
-        self.headers['Authorization'] = f'Bearer {auth_data["access_token"]}'
-        self.feed_url = auth_data['_links']['events']['href']
-        self.query_url = auth_data['_links']['ghostflame']['href']
-        self.bills_url = auth_data['_links']['bills_summary']['href']
+        self._save_auth_data(response)
+
+    def authenticate_with_cert(self, cpf: str, password: str, cert_path: str):
+        self.client.set_cert(cert_path)
+        url = self.discovery.get_app_url('token')
+        payload = {
+            'grant_type': 'password',
+            'client_id': 'legacy_client_id',
+            'client_secret': 'legacy_client_secret',
+            'login': cpf,
+            'password': password
+        }
+
+        response = self.client.post(url, json=payload)
+
+        self._save_auth_data(response)
+
+        return response.get('refresh_token')
+
+    def authenticate_with_refresh_token(self, refresh_token: str, cert_path: str):
+        self.client.set_cert(cert_path)
+
+        url = self.discovery.get_app_url('token')
+        payload = {
+            'grant_type': 'refresh_token',
+            'client_id': 'legacy_client_id',
+            'client_secret': 'legacy_client_secret',
+            'refresh_token': refresh_token,
+        }
+
+        response = self.client.post(url, json=payload)
+
+        self._save_auth_data(response)
 
     def get_card_feed(self):
-        request = requests.get(self.feed_url, headers=self.headers)
-        return json.loads(request.content.decode('utf-8'))
+        return self.client.get(self.feed_url)
 
     def get_card_statements(self):
         feed = self.get_card_feed()
         return list(filter(lambda x: x['category'] == 'transaction', feed['events']))
 
     def get_bills(self):
-        request = requests.get(self.bills_url, headers=self.headers)
-        return json.loads(request.content.decode('utf-8'))['bills']
+        request = self.client.get(self.bills_url)
+        return request['bills']
 
     def get_bill_details(self, bill):
-        request = requests.get(bill['_links']['self']['href'], headers=self.headers)
-        return json.loads(request.content.decode('utf-8'))
+        return self.client.get(bill['_links']['self']['href'])
 
     def get_account_feed(self):
         data = self._make_graphql_request('account_feed')

pynubank/utils/__init__.py

@@ -0,0 +1,91 @@
+import OpenSSL
+import requests
+from OpenSSL.crypto import X509
+
+from pynubank import NuException, NuRequestException
+from pynubank.utils.discovery import Discovery
+from pynubank.utils.http import HttpClient
+
+
+class CertificateGenerator:
+
+    def __init__(self, login, password, device_id):
+        self.login = login
+        self.password = password
+        self.device_id = device_id
+        self.encrypted_code = None
+        self.key1 = self._generate_key()
+        self.key2 = self._generate_key()
+        discovery = Discovery(HttpClient())
+        self.url = discovery.get_app_url('gen_certificate')
+
+    def request_code(self) -> str:
+        response = requests.post(self.url, json=self._get_payload())
+
+        if response.status_code != 401:
+            raise NuException('Authentication code request failure.')
+
+        parsed = self._parse_authenticate_headers(response.headers.get('WWW-Authenticate'))
+        self.encrypted_code = parsed.get('device-authorization_encrypted-code')
+
+        return parsed['sent-to']
+
+    def exchange_certs(self, code: str):
+        if not self.encrypted_code:
+            raise NuException('No encrypted code found. Did you call `request_code` before exchanging certs ?')
+
+        payload = self._get_payload()
+        payload['code'] = code
+        payload['encrypted-code'] = self.encrypted_code
+
+        response = requests.post(self.url, json=payload)
+
+        if response.status_code != 200:
+            raise NuRequestException(response)
+
+        data = response.json()
+
+        cert1 = self._parse_cert(data['certificate'])
+        cert2 = self._parse_cert(data['certificate_crypto'])
+
+        return self._gen_cert(self.key1, cert1), self._gen_cert(self.key2, cert2)
+
+    def _get_payload(self):
+        return {
+            'login': self.login,
+            'password': self.password,
+            'public_key': self._get_public_key(self.key1),
+            'public_key_crypto': self._get_public_key(self.key2),
+            'model': 'PyNubank Client',
+            'device_id': self.device_id
+        }
+
+    def _parse_cert(self, content: str) -> X509:
+        return OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, content.encode())
+
+    def _gen_cert(self, key, cert):
+        p12 = OpenSSL.crypto.PKCS12()
+        p12.set_privatekey(key)
+        p12.set_certificate(cert)
+
+        return p12
+
+    def _generate_key(self):
+        key = OpenSSL.crypto.PKey()
+        key.generate_key(OpenSSL.crypto.TYPE_RSA, 2048)
+
+        return key
+
+    def _get_public_key(self, key) -> str:
+        return OpenSSL.crypto.dump_publickey(OpenSSL.crypto.FILETYPE_PEM, key).decode()
+
+    def _parse_authenticate_headers(self, header_content: str) -> dict:
+        chunks = header_content.split(',')
+        parsed = {}
+        for chunk in chunks:
+            key, value = chunk.split('=')
+            key = key.strip().replace(' ', '_')
+            value = value.replace('"', '')
+            parsed[key] = value
+
+        return parsed

pynubank/utils/certificate_generator.py

@@ -0,0 +1,35 @@
+from pynubank.exception import NuException
+from pynubank.utils.http import HttpClient
+
+DISCOVERY_URL = 'https://prod-s0-webapp-proxy.nubank.com.br/api/discovery'
+DISCOVERY_APP_URL = 'https://prod-s0-webapp-proxy.nubank.com.br/api/app/discovery'
+
+
+class Discovery:
+    _headers = {
+        'Content-Type': 'application/json'
+    }
+
+    proxy_list_url: dict
+    proxy_list_app_url: dict
+
+    def __init__(self, client: HttpClient):
+        self.client = client
+        self._update_proxy_urls()
+
+    def get_url(self, name: str) -> str:
+        return self._get_url(name, self.proxy_list_url)
+
+    def get_app_url(self, name: str) -> str:
+        return self._get_url(name, self.proxy_list_app_url)
+
+    def _update_proxy_urls(self):
+        self.proxy_list_url = self.client.get(DISCOVERY_URL)
+        self.proxy_list_app_url = self.client.get(DISCOVERY_APP_URL)
+
+    def _get_url(self, name: str, target: dict) -> str:
+        url = target.get(name)
+        if not url:
+            raise NuException(f'There is no URL discovered for {name}')
+
+        return url