Merge branch 'dev_theo'

Author: ecrou-exact

website/db_class/db.py

@@ -134,3 +134,52 @@ def to_share(self):
             "share_url": f"http://cti-transmute.org/convert/share/{self.uuid}",
             "detail_url": f"http://cti-transmute.org/convert/detail/{self.id}"
         }
+
+class ConvertHistory(db.Model):
+    __tablename__ = "convert_history"
+
+    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
+    user_id = db.Column(db.Integer, nullable=True)
+
+    convert_id = db.Column(
+        db.Integer,
+        db.ForeignKey("convert.id", ondelete="CASCADE"),
+        nullable=False
+    )
+
+    version = db.Column(db.Integer, nullable=False)
+    uuid = db.Column(db.String(36), index=True, nullable=False)
+    status = db.Column(db.String(20), nullable=False) # pending , accepted , rejected
+    public = db.Column(db.Boolean, default=False, index=True)   # able to share with the community
+
+    input_text = db.Column(db.Text, nullable=True)
+
+    old_output_text = db.Column(db.Text, nullable=True)
+    new_output_text = db.Column(db.Text, nullable=True)
+
+    # Metadata
+    created_at = db.Column(db.DateTime, index=True)
+    comment = db.Column(db.Text, nullable=True)
+
+    # Relationship
+    convert = db.relationship("Convert", backref=db.backref(
+        "history",
+        lazy=True,
+        cascade="all, delete-orphan"
+    ))
+
+    def to_json(self):
+        return {
+            "id": self.id,
+            "convert_id": self.convert_id,
+            "version": self.version,
+            "uuid": self.uuid,
+            "input_text": self.input_text,
+            "old_output_text": self.old_output_text,
+            "new_output_text": self.new_output_text,
+            "created_at": self.created_at.strftime('%Y-%m-%d %H:%M') if self.created_at else None,
+            "comment": self.comment,
+            "status": self.status,
+            "public": self.public
+        }
+    

website/migrations/versions/d7053174c298_.py

@@ -0,0 +1,53 @@
+"""empty message
+
+Revision ID: d7053174c298
+Revises: a4ecea6f4ec3
+Create Date: 2025-11-26 10:12:07.474094
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'd7053174c298'
+down_revision = 'a4ecea6f4ec3'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('convert_history',
+    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+    sa.Column('user_id', sa.Integer(), nullable=True),
+    sa.Column('convert_id', sa.Integer(), nullable=False),
+    sa.Column('version', sa.Integer(), nullable=False),
+    sa.Column('uuid', sa.String(length=36), nullable=False),
+    sa.Column('status', sa.String(length=20), nullable=False),
+    sa.Column('public', sa.Boolean(), nullable=True),
+    sa.Column('input_text', sa.Text(), nullable=True),
+    sa.Column('old_output_text', sa.Text(), nullable=True),
+    sa.Column('new_output_text', sa.Text(), nullable=True),
+    sa.Column('created_at', sa.DateTime(), nullable=True),
+    sa.Column('comment', sa.Text(), nullable=True),
+    sa.ForeignKeyConstraint(['convert_id'], ['convert.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    with op.batch_alter_table('convert_history', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_convert_history_created_at'), ['created_at'], unique=False)
+        batch_op.create_index(batch_op.f('ix_convert_history_public'), ['public'], unique=False)
+        batch_op.create_index(batch_op.f('ix_convert_history_uuid'), ['uuid'], unique=False)
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('convert_history', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_convert_history_uuid'))
+        batch_op.drop_index(batch_op.f('ix_convert_history_public'))
+        batch_op.drop_index(batch_op.f('ix_convert_history_created_at'))
+
+    op.drop_table('convert_history')
+    # ### end Alembic commands ###

website/version

@@ -0,0 +1 @@
+1.1.0

website/web/convert/convert.py

@@ -1,4 +1,5 @@
 # website/web/convert/views.py
+import io
 import json
 from flask import Blueprint, jsonify, redirect, render_template, request, flash, url_for
 from flask_login import current_user, login_required
@@ -128,7 +129,7 @@ def stix_to_misp():
             if form.single_event.data:
                 params["single_event"] = ""
 
-            print(params)
+            #print(params)
 
             try:
                 response = requests.post(
@@ -439,3 +440,248 @@ def share_convert():
         return redirect(url_for("convert.history"))
 
     return render_template("convert/detail.html", convert=convert)
+
+
+
+###########################
+#   Refresh a conversion  #
+###########################
+
+@convert_blueprint.route("/refresh/<string:uuid>", methods=['GET', 'POST'])
+def refresh(uuid):
+
+    convert_obj = ConvertModel.get_convert_by_uuid(uuid)
+
+    if not convert_obj:
+        flash("Conversion not found.", "danger")
+        return redirect(url_for("convert.history"))
+
+    # Choose which WTForm to use
+    if convert_obj.conversion_type == "MISP_TO_STIX":
+        print("Using MISP to STIX form")
+        form = mispToStixParamForm()
+    elif convert_obj.conversion_type == "STIX_TO_MISP":
+        print("Using STIX to MISP form")
+        form = stixToMispParamForm()
+    else:
+        flash("Unsupported conversion type.", "danger")
+        return redirect(url_for("convert.history"))
+
+    # Prefill form (GET)
+    if request.method == "GET":
+        form.name.data = convert_obj.name
+        form.description.data = convert_obj.description
+        form.public.data = convert_obj.public
+
+    result = None
+    diff = None
+    error = None
+
+    if form.validate_on_submit():
+
+        # Call the generic dispatcher
+        new_output, is_identical, error_msg = ConvertModel.reconvert_conversion(convert_obj, form)
+
+        if error_msg:
+            error = error_msg
+            flash(error_msg, "danger")
+        else:
+            if not is_identical:
+                flash("Conversion re-executed successfully! Changes detected.", "success")
+                result = new_output
+                diff = "The new conversion result is DIFFERENT from the previous one."
+            else:
+                flash("Conversion re-executed successfully! No changes detected.", "success")
+                result = new_output
+                diff = "The new conversion result is IDENTICAL to the previous one."
+
+    return render_template(
+        "convert/refresh.html",
+        form=form,
+        convert_obj=convert_obj,
+        result=result,
+        diff=diff,
+        error=error,
+        filename=f"{convert_obj.name}_refresh.json"
+    )
+
+
+# get_history
+
+@convert_blueprint.route("/get_history", methods=['GET'])
+def get_history():
+    id = request.args.get('id', 1, type=int)
+    if id:
+        convert_obj = ConvertModel.get_convert(id)
+        if convert_obj:
+            latest_history = ConvertModel.get_history_list(convert_obj.id)
+            if latest_history:
+                return {
+                    "success": True, 
+                    "history_convert": [h.to_json() for h in latest_history],
+                    "message": "New convert found", 
+                    "toast_class" : "success"
+                    }, 200
+            return {
+                "success": True, 
+                "message": "No conversion history found for this convert", 
+                "toast_class" : "danger"
+                }, 200
+        return {
+            "success": False, 
+            "message": "No convert found for this id", 
+            "toast_class" : "danger"
+            }, 500
+    return {
+        "success": False, 
+        "message": "No id provided", 
+        "toast_class" : "danger"
+        }, 500
+
+
+
+
+@convert_blueprint.route("/get_new_convert", methods=['GET'])
+def get_new_convert():
+    """Get the new convert after a refresh to show the difference"""
+    id = request.args.get('id', 1, type=int)
+    if id:
+        convert_obj = ConvertModel.get_convert(id)
+        if convert_obj:
+            latest_history = ConvertModel.get_latest_history_list(convert_obj.id)
+            if latest_history:
+                return {
+                    "success": True, 
+                    "history_convert": [h.to_json() for h in latest_history],
+                    "message": "New convert found", 
+                    "toast_class" : "success"
+                    }, 200
+            return {
+                "success": True, 
+                "message": "No conversion history found for this convert", 
+                "toast_class" : "danger"
+                }, 200
+        return {
+            "success": False, 
+            "message": "No convert found for this id", 
+            "toast_class" : "danger"
+            }, 500
+    return {
+        "success": False, 
+        "message": "No id provided", 
+        "toast_class" : "danger"
+        }, 500
+
+
+@convert_blueprint.route("/history_action", methods=['GET'])
+def history_action():
+    """Handle actions related to conversion history"""
+    action = request.args.get('action')
+    history_id = request.args.get('history_id', type=int)
+    convert_id = request.args.get('convert_id', type=int)
+
+    if current_user.is_anonymous():
+        return {
+            "success": False,
+            "message": "You must be logged in to perform this action.",
+            "toast_class": "danger"
+        }, 401
+    if not current_user.is_admin() and current_user.id != ConvertModel.get_convert(convert_id).user_id:
+        return {
+            "success": False,
+            "message": "You do not have permission to perform this action.",
+            "toast_class": "danger"
+        }, 403
+    else:
+        # --- Handle Conversion History Actions (Accept/Reject) ---
+        if history_id and action in ["accept", "reject"]:
+            
+            if action == "accept":
+                success = ConvertModel.accept_history(history_id)
+            elif action == "reject":
+                success = ConvertModel.reject_history(history_id)
+            success = True # Replace with actual database call
+            
+            if success:
+                flash("Convert history updated", "success")
+                return {
+                    "success": True,
+                    "message": f"History entry {history_id} {action}ed successfully.",
+                    "toast_class": "success"
+                }, 200
+            return {
+                "success": False,
+                "message": f"Failed to {action} history entry {history_id}.",
+                "toast_class": "danger"
+            }, 500
+
+    # --- Default Error ---
+    return {
+        "success": False,
+        "message": "Invalid action or missing parameters (history_id or convert_id).",
+        "toast_class": "danger"
+    }, 400
+
+
+@convert_blueprint.route("/difference/<int:id>", methods=['GET'])
+def difference(id):
+    """Show the difference between two convert versions"""
+    convert_obj_history = ConvertModel.get_convert_history_by_id(id)
+    if not convert_obj_history:
+        flash("Conversion not found.", "danger")
+        return redirect(url_for("convert.history"))
+
+    convert_obj = ConvertModel.get_convert(convert_obj_history.convert_id)
+    if not convert_obj:
+        flash("Conversion not found.", "danger")
+        return redirect(url_for("convert.history"))
+    
+    if convert_obj.public == False:
+        if current_user.is_anonymous():
+            flash("You must be logged in to view this convert if you are the owner of this convert.", "warning")
+            return redirect(url_for("account.login"))  
+
+        if current_user.id != convert_obj.user_id and not current_user.is_admin():
+            flash("You do not have permission to view this convert.", "danger")
+            return redirect(url_for("convert.history"))
+        
+        return render_template(
+            "convert/compare_version/difference.html",
+            old_result=convert_obj_history.old_output_text,
+            new_result=convert_obj_history.new_output_text,
+            convert_obj=convert_obj,
+            history_id=convert_obj_history.id
+        )
+    else:
+        return render_template(
+            "convert/compare_version/difference.html",
+            old_result=convert_obj_history.old_output_text,
+            new_result=convert_obj_history.new_output_text,
+            convert_obj=convert_obj,
+            history_id=convert_obj_history.id
+        )
+
+# get_history_details
+@convert_blueprint.route("/get_history_details", methods=['GET'])
+def get_history_details():
+    """Get the details of a convert history entry"""
+    history_id = request.args.get('history_id', type=int)
+    if history_id:
+        convert_history = ConvertModel.get_convert_history_by_id(history_id)
+        if convert_history:
+            return {
+                "success": True, 
+                "history": convert_history.to_json(),
+                "message": "Convert history found", 
+                "toast_class" : "success"
+                }, 200
+        return {
+            "success": False, 
+            "message": "No convert history found for this id", 
+            "toast_class" : "danger"
+            }, 500
+    return {
+        "success": False, 
+        "message": "No history_id provided", 
+        "toast_class" : "danger"
+        }, 500