#!/usr/bin/env python
# -*- coding: utf_8 -*-
import os
import io
import time
import datetime
from werkzeug.routing import BaseConverter
from werkzeug.utils import secure_filename
from flask import Flask, jsonify, request
from flask import render_template
import core.settings as settings
import core.utils as utils
from core.scanner import general_code_analysis
from database import db_session
from models import Results
class RegexConverter(BaseConverter):
def __init__(self, url_map, *items):
super(RegexConverter, self).__init__(url_map)
self.regex = items[0]
app = Flask(__name__)
app.url_map.converters['regex'] = RegexConverter
app.config['DEBUG'] = settings.DEBUG
app.config['UPLOAD_FOLDER'] = settings.UPLOAD_FOLDER
app.config['SQLALCHEMY_DATABASE_URI'] = settings.SQLALCHEMY_DATABASE_URI
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
@app.template_filter('slugify')
def _slugify(string):
if not string:
return ""
return utils.slugify(string)
@app.context_processor
def _year():
return dict(year=str(utils.year()))
@app.template_filter('js_escape')
def _js_escape(string):
if not string:
return ""
return utils.js_escape(string)
@app.teardown_appcontext
def shutdown_session(exception=None):
db_session.remove()
@app.route('/', methods=['GET'])
def index():
context = {'title': 'NodeJSScan V2'}
return render_template("index.html", **context)
@app.route('/upload/', methods=['POST'])
def upload():
"""Upload Zipped Source"""
if 'file' in request.files:
filen = request.files['file']
_, extension = os.path.splitext(filen.filename.lower())
# Check for Valid ZIP
if (filen and
filen.filename and
extension in settings.UPLD_ALLOWED_EXTENSIONS and
filen.mimetype in settings.UPLD_MIME
):
filename = secure_filename(filen.filename)
# Make upload dir
if not os.path.exists(settings.UPLOAD_FOLDER):
os.makedirs(settings.UPLOAD_FOLDER)
# Save file
zip_file = os.path.join(app.config['UPLOAD_FOLDER'], filename)
filen.save(zip_file)
# Get zip hash
get_zip_hash = utils.gen_sha256_file(zip_file)
# check if already scanned
res = Results.query.filter(Results.scan_hash == get_zip_hash)
if not res.count():
# App analysis dir
app_dir = os.path.join(
app.config['UPLOAD_FOLDER'], get_zip_hash + "/")
# Make app analysis dir
if not os.path.exists(app_dir):
os.makedirs(app_dir)
# Unzip
utils.unzip(zip_file, app_dir)
# Do scan
scan_results = general_code_analysis([app_dir])
print "[INFO] Static Analysis Completed!"
_, sha2_hashes, hash_of_sha2 = utils.gen_hashes([app_dir])
tms = datetime.datetime.fromtimestamp(
time.time()).strftime('%Y-%m-%d %H:%M:%S')
# Save Result
print "[INFO] Saving Scan Results!"
res_db = Results(get_zip_hash,
[app_dir],
sha2_hashes,
hash_of_sha2,
scan_results['sec_issues'],
scan_results['good_finding'],
scan_results['missing_sec_header'],
scan_results['files'],
scan_results['total_count'],
scan_results['vuln_count'],
[],
[],
tms,
)
db_session.add(res_db)
db_session.commit()
return jsonify({"status": "success", "url": "result/" + get_zip_hash})
return jsonify({"status": "error", "desc": "Upload Failed!"})
@app.route('/dashboard/', methods=['GET'])
def dashboard():
"""Display dashboard"""
context = {}
res_shas = []
ress = Results.query.all()
for res in ress:
locations = res.locations
res_shas.append({"scan_hash": res.scan_hash,
"locations": locations,
"timestamp": str(res.timestamp),
}
)
context = {
'title': "Scan Dashboard",
'scan_details': res_shas,
}
return render_template("dashboard.html", **context)
@app.route('/result/