Baselined from internal Repository

last_commit:05893928151678d266b454967c7ca62d1f10e6ef

Author: GVE Devnet Admin

Dockerfile

@@ -23,7 +23,7 @@ preventing unauthorized access to sensitive information.
 ### Backend
 #### Retrieve Duo Credentials 
 1. Follow the instructions under 'First Steps' to get your Duo integration key, secret key and API hostname: https://duo.com/docs/authapi.
-2. Clone this repository with git clone [repository name]
+2. Clone this repository with git clone `git clone https://github.com/gve-sw/gve_devnet_duo_manual_push_auth.git`.
 3. Create and update .env (see below)
 4. Proceed to 'usage' section.
 
@@ -36,32 +36,41 @@ open .env
 ```
 Copy/Paste the .env variables and update accordingly:
 ```script
-DUO_IKEY=YOUR_DUO_INTEGRATION_KEY
-DUO_SKEY=YOUR_DUO_SECRET_KEY
-DUO_API_URL=YOUR_API_HOSTNAME
 APP_NAME='Duo Manual Push Authentication'
 APP_VERSION=1.0
 LOGGER_LEVEL=DEBUG
+DUO_API_URL=YOUR_API_HOSTNAME
+DUO_IKEY=YOUR_DUO_INTEGRATION_KEY
+DUO_SKEY=YOUR_DUO_SECRET_KEY
+DUO_ADMIN_API_URL=YOUR_ADMIN_API_HOSTNAME
+DUO_ADMIN_IKEY=YOUR_DUO_ADMIN_INTEGRATION_KEY
+DUO_ADMIN_SKEY=YOUR_DUO_ADMIN_SECRET_KEY
 ```
-* Note: APP_NAME and APP_VERSION are for auto generated FastAPI docs at uvicorn_running_url/docs (i.e. http://127.0.0.1:8000/docs )
+* Note: APP_NAME and APP_VERSION are for auto-generated FastAPI docs at uvicorn_running_url/docs (i.e. http://127.0.0.1:8000/docs )
 
 ## Usage
-### 1. Start Frontend
+### With Docker
+#### 1. To build and start the containers, run: ``` docker-compose up --build ```
+#### 2. Navigate to URL: ``` http://localhost:5173/ ```
+#### 3. To stop the containers and remove them along with their network, run: ``` docker-compose down ```
+
+### Without Docker
+#### 1. Start Frontend
 Terminal 1:
 ```script
 cd frontend
 npm install
 npm run dev
 ```
 
-### 2. Start Backend
+#### 2. Start Backend
 Terminal 2:
 ```scipt
 cd backend
 uvicorn main:app --reload
 ```
 
-### 3. Navigate to Frontend URL:
+#### 3. Navigate to Frontend URL:
 ```script
 http://localhost:5173/
 ```

README.md

@@ -0,0 +1,21 @@
+# Backend/Dockerfile
+# Use an official Python runtime as a parent image
+FROM python:3.11-slim-buster
+
+# Set the working directory in the container
+WORKDIR /app
+
+# Copy the current directory contents into the container at /app
+COPY . .
+
+# Install any needed packages specified in requirements.txt
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Make port 8000 available to the world outside this container
+EXPOSE 8000
+
+# Define environment variable
+# ENV NAME World
+
+# Run app.py when the container launches
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

backend/Dockerfile

@@ -18,7 +18,6 @@
 from pydantic import field_validator
 from pydantic_settings import BaseSettings
 
-
 # Dynamically locate the .env file and handle cases where it might not exist.
 try:
     env_path = pathlib.Path(__file__).parents[0] / '.env'
@@ -43,6 +42,9 @@ class Config(BaseSettings):
     DUO_IKEY: str
     DUO_SKEY: str
 
+    DUO_ADMIN_API_URL: str
+    DUO_ADMIN_IKEY: str
+    DUO_ADMIN_SKEY: str
 
     @field_validator('DUO_API_URL', mode='before')
     def validate_duo_api_url(cls, v):
@@ -51,4 +53,5 @@ def validate_duo_api_url(cls, v):
         return v
 
 
-config = Config()      # Create a single instance of Config()
+
+config = Config()  # Create a single instance of Config()

backend/config/config.py

@@ -20,19 +20,22 @@
 import requests
 from urllib.parse import urlparse
 from config.config import config
-
-# Access environment variables
-IKEY = config.DUO_IKEY
-SKEY = config.DUO_SKEY
-API_URL = config.DUO_API_URL
+import time
+from pprint import pprint
 
 
 class DuoAuthenticator:
-    def __init__(self, ikey, skey, api_url):
-        self.ikey = ikey
-        self.skey = skey
-        self.api_url = api_url
-        self.host = self.parse_hostname(api_url)
+    def __init__(self):
+        # Access environment variables
+        # For both Auth and Admin Duo API
+        self.auth_ikey = config.DUO_IKEY
+        self.auth_skey = config.DUO_SKEY
+        self.auth_api_url = config.DUO_API_URL
+        self.auth_host = self.parse_hostname(self.auth_api_url)
+        self.admin_ikey = config.DUO_ADMIN_IKEY
+        self.admin_skey = config.DUO_ADMIN_SKEY
+        self.admin_api_url = config.DUO_ADMIN_API_URL
+        self.admin_host = self.parse_hostname(self.admin_api_url)
 
     def parse_hostname(self, url):
         parsed_url = urlparse(url)
@@ -43,49 +46,109 @@ def parse_hostname(self, url):
         return host
 
     def generate_headers(self, method, path, params):
+        if path.startswith('/admin'):
+            ikey = self.admin_ikey
+            skey = self.admin_skey
+            host = self.admin_host
+        else:
+            ikey = self.auth_ikey
+            skey = self.auth_skey
+            host = self.auth_host
         now = email.utils.formatdate()
-        canon = [now, method.upper(), self.host.lower(), path]
+        canon = [now, method.upper(), host.lower(), path]
         args = []
         for key in sorted(params.keys()):
             val = params[key].encode("utf-8")
             args.append(
                 '%s=%s' % (urllib.parse.quote(key, '~'), urllib.parse.quote(val, '~')))
         canon.append('&'.join(args))
         canon = '\n'.join(canon)
-        sig = hmac.new(bytes(self.skey, encoding='utf-8'),
+        sig = hmac.new(bytes(skey, encoding='utf-8'),
                        bytes(canon, encoding='utf-8'), hashlib.sha1)
-        auth = '%s:%s' % (self.ikey, sig.hexdigest())
+        auth = '%s:%s' % (ikey, sig.hexdigest())
         return ('&'.join(args), {
             'Date': now,
             'Authorization': 'Basic %s' % base64.b64encode(bytes(auth, encoding="utf-8")).decode(),
             'Content-Type': 'application/x-www-form-urlencoded'
         })
 
-    def authenticate_user(self, user_email):
+    def authenticate_user(self, payload):
+        user_email = payload.get("email", "")
+        username = payload.get("username", "")
+        device = "auto"  # or extract from payload's devices array if needed
+
         uri = '/auth/v2/auth'
-        params = {'username': user_email, 'factor': 'push', 'device': 'auto', 'async': '1'}
+        params = {'username': username, 'factor': 'push', 'device': device, 'async': '1'}
         body, headers = self.generate_headers('POST', uri, params)
-        response = requests.post(f'https://{self.host}{uri}', headers=headers, data=body).json()
+
+        # Use self.auth_host instead of self.host
+        response = requests.post(f'https://{self.auth_host}{uri}', headers=headers, data=body).json()
+
         if response['stat'] != 'OK':
             return response
         return self.check_auth_status(response['response']['txid'])
 
     def check_auth_status(self, txid, timeout=60, interval=5):
-        import time
         uri = '/auth/v2/auth_status'
         params = {'txid': txid}
         start_time = time.time()
         while True:
             if time.time() - start_time > timeout:
                 return "Error: Timeout"
             args, headers = self.generate_headers('GET', uri, params)
-            result = requests.get(f'https://{self.host}{uri}?{args}', headers=headers).json()
+            result = requests.get(f'https://{self.auth_host}{uri}?{args}', headers=headers).json()
             if result['stat'] != 'OK':
                 return result
             if result['response']['result'] in ['allow', 'deny']:
                 return result['response']['result']
             time.sleep(interval)
 
+    def fetch_users(self):
+        uri = '/admin/v1/users'
+        params_f = '?limit={}&offset={}'
+        result = []
+        more = True
+        limit = '100'
+        offset = '0'
+        while more:
+            params = params_f.format(limit, offset)
+            body, headers = self.generate_headers('GET', uri, {'limit': limit, 'offset': offset})
+            response = requests.get(f'https://{self.admin_host}{uri}{params}', headers=headers).json()
+            if response['stat'] != 'OK':
+                return response
+            for user in response['response']:
+                if user['status'] != 'active' and user['status'] != 'bypass':
+                    continue
+                user_dict = {
+                    'username': user['username'],
+                    'fullname': user['realname'],
+                    'email': user['email'],
+                    'status': user['status'],
+                }
+                devices = []
+                for phone in user['phones']:
+                    if phone['activated']:
+                        try:
+                            phone['capabilities'].remove('auto')
+                        except ValueError:
+                            pass
+                        devices.append({
+                            'id': phone['phone_id'],
+                            'type': 'phone',
+                            'capabilities': phone['capabilities'],
+                            'model': phone['model'],
+                            'number': phone['number']
+                        })
+                user_dict['devices'] = devices
+                result.append(user_dict)
+            if response['metadata'].get('next_offset'):
+                pprint(response['metadata'])
+                offset = str(response['metadata'].get('next_offset'))
+            else:
+                pprint(response['metadata'])
+                more = False
+        return result
+
 
 # Instantiate the DuoAuthenticator
-duo_authenticator = DuoAuthenticator(IKEY, SKEY, API_URL)
+duo_authenticator = DuoAuthenticator()

backend/duo_app.py

@@ -10,29 +10,50 @@
 IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 or implied.
 """
-
 from fastapi import APIRouter, Request, Depends, HTTPException
+from typing import List, Optional
 from pydantic import BaseModel
 from logrr import logger_manager
 from duo_app import duo_authenticator
 
 router = APIRouter()
 
 
-class UserRequest(BaseModel):
-    email: str
+class Device(BaseModel):
+    id: str
+    type: str
+    capabilities: List[str]
+    model: Optional[str] = None
+    number: Optional[str] = None
+
+
+class User(BaseModel):
+    username: str
+    fullname: str
+    email: Optional[str] = None
+    status: str
+    devices: List[Device]
 
 
-# Using duo_app.py
 @router.post("/authenticate/")
-def authenticate(user_request: UserRequest):
+async def authenticate(user_request: User):
+    try:
+        user_request_dict = user_request.model_dump()  # Convert the user_request object to a dictionary
+        result = duo_authenticator.authenticate_user(user_request_dict)  # Pass the user_request_dict to the authenticate_user function
+        # Simulate authentication result for demonstration purposes (optional)
+        # result = "Authentication successful for user: " + user_request.username
+        return {"output": result}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))  # Log and handle the error
+
+
+@router.get("/users/")
+def users():
     try:
-        logger_manager.console.print('[orange1]Authenticating with Duo...[/orange1]')
-        result = duo_authenticator.authenticate_user(user_request.email)
-        # result = duo_authenticator.parse_hostname(config.DUO_API_URL)
-        logger_manager.console.print(f'Result {result}')
+        logger_manager.console.print('[orange1]Fetching users...[/orange1]')
+        result = duo_authenticator.fetch_users()
         return {"output": result}
     except Exception as e:
         # Log and handle the error
         logger_manager.console.print(f"[red]Error: {e}[/red]")
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=500, detail=str(e))

requirements.txt renamed to backend/requirements.txt

@@ -1,13 +1,22 @@
-version: "3.5"
+version: "3.8"
 
 services:
-  Duo_Manual_Push_Auth:
-    image: ghcr.io/gve-sw/gve_devnet_duo_manual_push_auth:latest
-    container_name: gve_devnet_duo_manual_push_auth:container
-    environment:
-      - SOME_VAR=""
+  frontend:
+    build:
+      context: ./frontend
+      dockerfile: Dockerfile
+    restart: always
     ports:
+      - "5173:5173"
+    environment:
+      - NODE_ENV=development
 
-    volumes:
-      - config.yaml:/main/config.yaml
-    restart: "always"
+  backend:
+    env_file:
+      - ./backend/config/.env
+    build:
+      context: ./backend
+      dockerfile: Dockerfile
+    restart: always
+    ports:
+      - "8000:8000"

backend/routes.py

@@ -0,0 +1,24 @@
+# Frontend/Dockerfile
+# Use an official Node runtime as a parent image
+FROM node:latest
+
+# Set the working directory in the container
+WORKDIR /usr/src/app
+
+# Copy package.json and package-lock.json
+COPY package*.json ./
+
+# Try a clean install
+RUN npm ci
+
+# Bundle app source
+COPY . .
+
+# Your app binds to port 5173, so use the EXPOSE instruction
+EXPOSE 5173
+
+# Define environment variable
+ENV NODE_ENV development
+
+# Run npm run dev when the container launches
+CMD ["npm", "run", "dev"]