Skip to content
Snippets Groups Projects
Commit bcfc4d58 authored by Benedikt von St. Vieth's avatar Benedikt von St. Vieth
Browse files

dont know whats that

parent b8bedee3
Branches
No related tags found
No related merge requests found
Pipeline #
import argparse
import logging
import os
import subprocess
import sys
import time
from datetime import datetime
from flask import Flask, render_template, request,\
send_from_directory as send_file
from flask_bootstrap import Bootstrap
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///logs.db'
Bootstrap(app)
db = SQLAlchemy(app)
command = ''
token = False
def _compare(token, auth_header):
logging.debug('Using own compare function for authorization token')
# not very performant, but posts are not very often
# and this code is not used in new python versions.
valid = True
if len(token) != len(auth_header):
return False
for iterator, item in enumerate(token):
if item != auth_header[iterator]:
valid = False
return valid
class WebhookCall(db.Model):
__tablename__ = 'webhookcall'
timestamp = db.Column(db.Integer, primary_key=True, unique=True)
repository = db.Column(db.String(20))
success = db.Column(db.Boolean)
results = db.relationship('WebhookCallResult',
backref=db.backref('webhookcall'))
def __init__(self, timestamp, repository, success):
self.timestamp = timestamp
self.repository = repository
self.success = success
def __repr__(self):
return '<WebhookCall %r>' % self.timestamp
def __gt__(self, other):
return True if int(self.timestamp) > int(other.timestamp) else False
def __lt__(self, other):
return True if int(self.timestamp) < int(other.timestamp) else False
class WebhookCallResult(db.Model):
__tablename__ = 'webhookcallresult'
id = db.Column(db.Integer, primary_key=True)
timestamp = db.Column(db.Integer, db.ForeignKey('webhookcall.timestamp'))
output = db.Column(db.String(1000))
def __init__(self, timestamp, output):
self.timestamp = timestamp
self.output = output
def __repr__(self):
return '<WebhookCallResult %r>' % self.id
def _restart():
global command
try:
output = subprocess.check_output(command.split())
except Exception as err:
output = 'Error {} while executing {}'.format(err, command)
return output
@app.route('/', methods=['POST'])
@app.route('/<hooking_repository>/', methods=['POST'])
def post_hook(hooking_repository=None):
global token
auth_header = request.headers.get('Authorization')
if not token:
pass
elif auth_header:
valid = False
for item in token.split(';'):
if compare_func(unicode(item), auth_header):
valid = True
if not valid:
return 'Access forbidden', 403
else:
return 'Access forbidden', 403
if hooking_repository:
if len(hooking_repository) < 20:
repository = hooking_repository
else:
repository = hooking_repository[0:20]
else:
repository = 'unknown'
current_time = int(time.time())
output = _restart()
success = False if 'Error' in output else True
output = output.strip()
item = WebhookCall.query.filter_by(timestamp=current_time).first()
if not item:
db.session.add(WebhookCall(current_time, repository, success))
db.session.add(WebhookCallResult(timestamp=current_time,
output=output))
else:
db.session.add(WebhookCallResult(timestamp=current_time,
output=output))
db.session.commit()
response = 'Executed restart action at {} with output:\n{}'\
.format(format_datetime(current_time), output)
if success:
return response, 200
else:
return response, 500
@app.route('/', methods=['GET'])
@app.route('/logs/', methods=['GET'])
def get_logs():
return render_template('logs.html', content=WebhookCall.query.all())
@app.route('/logs/<log_id>', methods=['GET'])
def get_log(log_id):
logs = WebhookCallResult.query.filter(
WebhookCallResult.timestamp == log_id).all()
if logs:
return render_template('log.html', timestamp=log_id, logs=logs)
else:
return '', 404
@app.route('/favicon.ico')
def favicon():
return send_file(os.path.join(app.root_path, 'static'), 'favicon.ico')
@app.template_filter('datetime')
def format_datetime(value):
return datetime.fromtimestamp(int(value))\
.strftime("%Y-%m-%d_%H-%M-%S")
activate_this = '/home/centos/webhook/bin/activate_this.py'
execfile(activate_this, dict(__file__=activate_this))
from webhook import app as application
if __name__ == '__main__':
application.run(host='0.0.0.0', port=8082, debug=True, use_reloader=False)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment