""" Django management command to register M-Pesa C2B callback URLs """ from django.core.management.base import BaseCommand, CommandError from django.conf import settings from payments.models import MpesaConfiguration from payments.services import MpesaService import requests import base64 class Command(BaseCommand): help = 'Register M-Pesa C2B callback URLs with Safaricom' def add_arguments(self, parser): parser.add_argument( '--shortcode', type=str, help='Business shortcode (paybill number)', ) parser.add_argument( '--consumer-key', type=str, help='M-Pesa Consumer Key', ) parser.add_argument( '--consumer-secret', type=str, help='M-Pesa Consumer Secret', ) parser.add_argument( '--validation-url', type=str, help='Validation URL (default: https://branchbusinessadvance.co.ke/payments/callback/validation/)', ) parser.add_argument( '--confirmation-url', type=str, help='Confirmation URL (default: https://branchbusinessadvance.co.ke/payments/callback/confirmation/)', ) parser.add_argument( '--response-type', type=str, default='Completed', choices=['Completed', 'Cancelled'], help='Response type for failed validation (default: Completed)', ) parser.add_argument( '--use-config', action='store_true', help='Use credentials from database configuration', ) parser.add_argument( '--production', action='store_true', help='Use production API endpoint', ) parser.add_argument( '--yes', action='store_true', help='Skip confirmation prompt (auto-confirm)', ) def handle(self, *args, **options): self.stdout.write(self.style.SUCCESS('=' * 70)) self.stdout.write(self.style.SUCCESS('M-Pesa C2B URL Registration')) self.stdout.write(self.style.SUCCESS('=' * 70)) # Determine credentials source if options['use_config']: self.stdout.write('Using credentials from database configuration...') config = MpesaConfiguration.get_active_config() if not config: raise CommandError('No active M-Pesa configuration found in database') shortcode = config.business_short_code consumer_key = config.consumer_key consumer_secret = config.consumer_secret validation_url = options.get('validation_url') or config.validation_url confirmation_url = options.get('confirmation_url') or config.confirmation_url response_type = options.get('response_type') or config.response_type is_production = config.environment == 'production' else: # Use command line arguments shortcode = options.get('shortcode') consumer_key = options.get('consumer_key') consumer_secret = options.get('consumer_secret') validation_url = options.get('validation_url') or 'https://branchbusinessadvance.co.ke/payments/callback/validation/' confirmation_url = options.get('confirmation_url') or 'https://branchbusinessadvance.co.ke/payments/callback/confirmation/' response_type = options.get('response_type') is_production = options.get('production', False) if not all([shortcode, consumer_key, consumer_secret]): raise CommandError( 'Please provide --shortcode, --consumer-key, and --consumer-secret, ' 'or use --use-config to load from database' ) # Display configuration self.stdout.write('') self.stdout.write(self.style.WARNING('Configuration:')) self.stdout.write(f' Environment: {"PRODUCTION" if is_production else "SANDBOX"}') self.stdout.write(f' Short Code: {shortcode}') self.stdout.write(f' Consumer Key: {consumer_key[:10]}...') self.stdout.write(f' Validation URL: {validation_url}') self.stdout.write(f' Confirmation URL: {confirmation_url}') self.stdout.write(f' Response Type: {response_type}') self.stdout.write('') # Confirm before proceeding if is_production and not options.get('yes'): self.stdout.write(self.style.ERROR('⚠️ WARNING: You are about to register URLs in PRODUCTION!')) confirm = input('Type "yes" to continue: ') if confirm.lower() != 'yes': self.stdout.write(self.style.WARNING('Registration cancelled.')) return try: # Step 1: Get access token self.stdout.write('Step 1: Generating access token...') access_token = self._get_access_token( consumer_key, consumer_secret, is_production ) self.stdout.write(self.style.SUCCESS(f'✓ Access token generated: {access_token[:20]}...')) # Step 2: Register URLs self.stdout.write('') self.stdout.write('Step 2: Registering callback URLs...') result = self._register_urls( access_token, shortcode, validation_url, confirmation_url, response_type, is_production ) # Display result self.stdout.write('') self.stdout.write(self.style.SUCCESS('=' * 70)) self.stdout.write(self.style.SUCCESS('✓ URL Registration Successful!')) self.stdout.write(self.style.SUCCESS('=' * 70)) self.stdout.write('') self.stdout.write('Response from M-Pesa:') for key, value in result.items(): self.stdout.write(f' {key}: {value}') self.stdout.write('') # Save to database if using config if options['use_config']: config.validation_url = validation_url config.confirmation_url = confirmation_url config.save() self.stdout.write(self.style.SUCCESS('✓ Configuration updated in database')) self.stdout.write('') self.stdout.write(self.style.SUCCESS('Next Steps:')) self.stdout.write('1. Test the integration by making a payment to your shortcode') self.stdout.write('2. Monitor the callback logs at /payments/callbacks/') self.stdout.write('3. Check transaction processing at /payments/transactions/') except Exception as e: self.stdout.write('') self.stdout.write(self.style.ERROR('=' * 70)) self.stdout.write(self.style.ERROR('✗ URL Registration Failed')) self.stdout.write(self.style.ERROR('=' * 70)) self.stdout.write(self.style.ERROR(f'Error: {str(e)}')) self.stdout.write('') raise CommandError(f'Failed to register URLs: {str(e)}') def _get_access_token(self, consumer_key, consumer_secret, is_production): """Generate M-Pesa access token""" if is_production: url = 'https://api.safaricom.co.ke/oauth/v1/generate?grant_type=client_credentials' else: url = 'https://sandbox.safaricom.co.ke/oauth/v1/generate?grant_type=client_credentials' # Create basic auth header credentials = f"{consumer_key}:{consumer_secret}" encoded_credentials = base64.b64encode(credentials.encode()).decode() headers = { 'Authorization': f'Basic {encoded_credentials}', 'Content-Type': 'application/json' } response = requests.get(url, headers=headers, timeout=30) response.raise_for_status() data = response.json() return data['access_token'] def _register_urls(self, access_token, shortcode, validation_url, confirmation_url, response_type, is_production): """Register callback URLs with M-Pesa""" if is_production: url = 'https://api.safaricom.co.ke/mpesa/c2b/v2/registerurl' else: url = 'https://sandbox.safaricom.co.ke/mpesa/c2b/v1/registerurl' headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json' } payload = { 'ShortCode': shortcode, 'ResponseType': response_type, 'ConfirmationURL': confirmation_url, 'ValidationURL': validation_url } response = requests.post(url, json=payload, headers=headers, timeout=30) # Log response for debugging try: response_data = response.json() if response.status_code != 200: raise Exception(f"M-Pesa API Error: {response_data}") except ValueError: # Not JSON response raise Exception(f"M-Pesa API returned non-JSON response: {response.text}") response.raise_for_status() return response_data