Initial AI Assistant implementation (fixes #8)

.github/workflows/debricked.yml

@@ -40,41 +40,18 @@ jobs:
     runs-on: ubuntu-latest
     if: ${{ (github.event_name == 'push') || (github.event_name == 'pull_request') || (github.event.inputs.runDebrickedScan == 'true') }}
     steps:
-      - name: Checkout
-        uses: actions/checkout@v4
-        with:
-          # Fetch at least the immediate parents so that if this is a pull request then we can checkout the head.
-          fetch-depth: 2
-      # Java is required to run the various Fortify utilities.
-      # Setup JDK 11 on host
-      - uses: actions/setup-java@v4
-        with:
-          distribution: 'temurin'
-          java-version: '11'
-      # Install Fortify (if required)
-      - name: Setup Fortify tools
-        uses: fortify/github-action/setup@v1.2.2
-        with:
-          export-path: true
-          fcli: latest
-      # Run debricked scan    
-      - name: Download Debricked CLI
-        run: curl -L https://github.com/debricked/cli/releases/latest/download/cli_linux_x86_64.tar.gz | tar -xz debricked
-      - name: Run Debricked scan  
-        run: ./debricked scan -r "${APP_NAME}" --access-token="${DEBRICKED_TOKEN}"  -e "*/**.lock" -e "**/build/classes/test/**" -e "**/target/classes/test-classes/**" .
+      # Docker example - uncomment to use Docker instead of GitHub Action
+      - uses: actions/checkout@v4
+      - uses: debricked/fingerprint@v4
+      - uses: debricked/resolve@v4
+      - uses: debricked/actions@v4
         env:
-          APP_NAME: ${{ env.DEFAULT_APP_NAME }}
           DEBRICKED_TOKEN: ${{ secrets.DEBRICKED_TOKEN }}
-
-scan_frontend:
-  stage: scan
-  image: debricked/cli:2-resolution-debian
-  script:
-    - debricked scan "$COMPONENT_DIR" -r ${DEBRICKED_REPO} -b ${DEBRICKED_BRANCH} -t ${DEBRICKED_TOKEN} --sbom CycloneDX --sbom-output gl-sbom-cdx.json
-  dependencies:
-    - build_frontend
-  allow_failure: true
-  artifacts:
-    reports:
-      cyclonedx:
-        - gl-sbom-cdx.json
+      # CLI example - uncomment to use CLI instead of GitHub Action
+      #- name: Download Debricked CLI
+      #  run: curl -L https://github.com/debricked/cli/releases/latest/download/cli_linux_x86_64.tar.gz | tar -xz debricked
+      #- name: Run Debricked scan
+      #  run: ./debricked scan -r "${APP_NAME}" --access-token="${DEBRICKED_TOKEN}"  -e "*/**.lock" -e "**/build/classes/test/**" -e "**/target/classes/test-classes/**" .
+      #  env:
+      #    APP_NAME: ${{ env.DEFAULT_APP_NAME }}
+      #    DEBRICKED_TOKEN: ${{ secrets.DEBRICKED_TOKEN }}

README.md

@@ -85,6 +85,14 @@ Scan Application (with OpenText Application Security)
 
 To carry out an OpenText Static Code Analyzer local scan, run the following:
 
+```
+sourceanalyzer -b iwa -clean
+sourceanalyzer -b iwa -python-path ".venv\\Lib\\site-packages" iwa
+sourceanalyzer -b iwa -scan
+```
+
+or you can use the Makefile target:
+
 ```
 make sast-scan
 ```
@@ -93,7 +101,7 @@ To carry out a OpenText ScanCentral SAST scan, run the following:
 
 ```
 fcli ssc session login
-scancentral package -o package.zip -bt none --python-virtual-env .venv -oss
+scancentral package -o package.zip -bt none --python-virtual-env .venv
 fcli sast-scan start --release "_YOURAPP_:_YOURREL_" -f package.zip --store curScan
 fcli sast-scan wait-for ::curScan::
 fcli ssc action run appversion-summary --av "_YOURAPP_:_YOURREL_" -fs "Security Auditor View" -f summary.md

iwa/__init__.py

@@ -67,7 +67,7 @@ def create_app(test_config=None):
     # enabled CORS on api routes
     cors = CORS(app, resources={r"/api/*": {"origins": "*"}})
 
-    # OpenAI configuration for Agent
+    # OpenAI configuration for assistant
     #app.config['OPENAI_API_KEY'] = os.getenv('OPENAI_API_KEY')
     #openai_extension.init_app(app)
 
@@ -107,8 +107,8 @@ def internal_error(error):
     app.register_blueprint(products_bp, url_prefix='/products')
     from iwa.blueprints.cart.cart_routes import cart_bp
     app.register_blueprint(cart_bp, url_prefix='/cart')
-    #from iwa.agent.routes import agent_bp
-    #app.register_blueprint(agent_bp)
+    from iwa.blueprints.assistant.assistant_routes import assistant_bp
+    app.register_blueprint(assistant_bp, url_prefix='/assistant')
     from iwa.blueprints.insecure.insecure_routes import insecure_bp
     app.register_blueprint(insecure_bp, url_prefix='/insecure')
 

iwa/blueprints/assistant/__init__.py

@@ -0,0 +1,24 @@
+"""
+        InsecureWebApp - an insecure Python/Flask Web application
+
+        Copyright (C) 2024-2025  Kevin A. Lee (kadraman)
+
+        This program is free software: you can redistribute it and/or modify
+        it under the terms of the GNU General Public License as published by
+        the Free Software Foundation, either version 3 of the License, or
+        (at your option) any later version.
+
+        This program is distributed in the hope that it will be useful,
+        but WITHOUT ANY WARRANTY; without even the implied warranty of
+        MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+        GNU General Public License for more details.
+
+        You should have received a copy of the GNU General Public License
+        along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+
+from flask import Blueprint
+
+assistant_bp = Blueprint("assistant", __name__, template_folder='templates', static_folder='static')
+
+from iwa.blueprints.assistant import assistant_routes

iwa/blueprints/assistant/assistant_routes.py

@@ -0,0 +1,169 @@
+"""
+        InsecureWebApp - an insecure Python/Flask Web application
+
+        Copyright (C) 2024-2025  Kevin A. Lee (kadraman)
+
+        This program is free software: you can redistribute it and/or modify
+        it under the terms of the GNU General Public License as published by
+        the Free Software Foundation, either version 3 of the License, or
+        (at your option) any later version.
+
+        This program is distributed in the hope that it will be useful,
+        but WITHOUT ANY WARRANTY; without even the implied warranty of
+        MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+        GNU General Public License for more details.
+
+        You should have received a copy of the GNU General Public License
+        along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+import logging
+import os
+import time
+
+from flask import render_template, request, jsonify, session
+from openai import OpenAI
+
+from iwa.blueprints.assistant import assistant_bp
+
+
+logger = logging.getLogger(__name__)
+
+# Initialize the OpenAI client
+api_key = os.getenv("OPENAI_API_KEY")
+if api_key:
+    logger.debug("OPENAI_API_KEY: %s", api_key)
+    logger.debug("AI Assistant functionality is enabled.")
+    client = OpenAI(api_key=api_key)
+else:
+    logger.debug("OpenAI API key not found. Assistant functionality is disabled.")
+
+# Initialize the assistant and thread globally
+assistant_id = ""
+thread_id = ""
+
+# Define a global chat history
+chat_history = [
+    {"role": "system", "content": "You are a helpful assistant."},
+]
+
+
+def create_assistant():
+    global assistant_id
+    if assistant_id == "":
+        my_assistant = client.beta.assistants.create(
+            instructions="""You are a helpful medical assistant. You can help with medical questions and provide 
+                information on various health topics. If you are unsure about a question, please advise the user to 
+                consult a healthcare professional. You can also help with scheduling appointments, providing information on medications,
+                and answering general health inquiries. Always prioritize user safety and confidentiality.
+            """,
+            name="MyMedicalAssistant",
+            model="gpt-3.5-turbo",
+            tools=[{"type": "file_search"}],
+        )
+        assistant_id = my_assistant.id
+    else:
+        my_assistant = client.beta.assistants.retrieve(assistant_id)
+        assistant_id = my_assistant.id
+
+    return my_assistant
+
+
+def create_thread():
+    global thread_id
+    if thread_id == "":
+        thread = client.beta.threads.create()
+        thread_id = thread.id
+    else:
+        thread = client.beta.threads.retrieve(thread_id)
+        thread_id = thread.id
+
+    return thread
+
+
+@assistant_bp.route('/', methods=['GET', 'POST'])
+def index():
+    """assistant page."""
+    if not api_key:
+        session["assistant_enabled"] = False
+        message = "The AI Assistant is not currently available. Please try again later!"
+    else:
+        session["assistant_enabled"] = True
+        message = "The AI Assistant is ready to help you!"
+    return render_template("assistant/index.html", message=message)
+
+@assistant_bp.route('/chat', methods=['POST'])
+def chat():
+    content = request.json["message"]
+    chat_history.append({"role": "user", "content": content})
+
+    # Send the message to the assistant
+    message_params = {"thread_id": thread_id, "role": "user", "content": content}
+
+    thread_message = client.beta.threads.messages.create(**message_params)
+
+    # Run the assistant
+    run = client.beta.threads.runs.create(
+        thread_id=thread_id, assistant_id=assistant_id
+    )
+    # Wait for the run to complete and get the response
+    while run.status != "completed":
+        time.sleep(0.5)
+        run = client.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id)
+
+    response = client.beta.threads.messages.list(thread_id).data[0]
+
+    text_content = None
+
+    # Iterate through the content objects to find the first text content
+    for content in response.content:
+        if content.type == "text":
+            text_content = content.text.value
+            break  # Exit the loop once the first text content is found
+
+    # Check if text content was found
+    if text_content:
+        chat_history.append({"role": "assistant", "content": text_content})
+        return jsonify(success=True, message=text_content)
+    else:
+        # Handle the case where no text content is found
+        return jsonify(success=False, message="No text content found")
+
+
+@assistant_bp.route('/reset', methods=['POST'])
+def reset_chat():
+    global chat_history
+    chat_history = [{"role": "system", "content": "You are a helpful assistant."}]
+
+    global thread_id
+    thread_id = ""
+    create_thread()
+    return jsonify(success=True)
+
+@assistant_bp.route("/get_ids", methods=["GET"])
+def get_ids():
+    return jsonify(assistant_id=assistant_id, thread_id=thread_id)
+
+@assistant_bp.route("/get_messages", methods=["GET"])
+def get_messages():
+    if thread_id != "":
+        thread_messages = client.beta.threads.messages.list(thread_id, order="asc")
+        messages = [
+            {
+                "role": msg.role,
+                "content": msg.content[0].text.value,
+            }
+            for msg in thread_messages.data
+        ]
+        return jsonify(success=True, messages=messages)
+    else:
+        return jsonify(success=False, message="No thread ID")
+
+@assistant_bp.before_request
+def before_request():
+    if api_key:
+        create_assistant()
+        create_thread()
+
+@assistant_bp.after_request
+def after_request(response):
+    return response