router.py (view raw)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# vim: set fileencoding=utf-8 :
__author__ = "Peter Molnar"
__copyright__ = "Copyright 2017, Peter Molnar"
__license__ = "GPLv3"
__version__ = "2.0"
__maintainer__ = "Peter Molnar"
__email__ = "hello@petermolnar.eu"
__status__ = "Production"
"""
silo archiver module of NASG
Copyright (C) 2017 Peter Molnar
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, write to the Free Software Foundation,
Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
"""
from sanic import Sanic
import sanic.response
#from sanic.log import log as logging
import logging
import validators
import urllib.parse
import shared
import envelope
import socket
if __name__ == '__main__':
#logging_format = "[%(asctime)s] %(process)d-%(levelname)s "
#logging_format += "%(module)s::%(funcName)s():l%(lineno)d: "
#logging_format += "%(message)s"
#logging.basicConfig(
#format=logging_format,
#level=logging.DEBUG
#)
#log = logging.getLogger()
# log_config=None prevents creation of access_log and error_log files
# since I'm running this from systemctl it already goes into syslog
app = Sanic('router')
#app = Sanic('router', log_config=None)
# this is ok to be read-only
sdb = shared.SearchDB()
@app.route("/oauth1", methods=["GET"])
async def oauth1(request):
token = request.args.get('oauth_token')
verifier = request.args.get('oauth_verifier')
logging.info("incoming oauth request: token was %s ; verifier was %s", token, verifier)
tokendb = shared.TokenDB()
tokendb.update_token(
token,
verifier=verifier
)
return sanic.response.text("OK", status=200)
@app.route("/search", methods=["GET"])
async def search(request):
query = request.args.get('s')
r = sdb.html(query)
response = sanic.response.html(r, status=200)
return response
@app.route("/micropub", methods=["POST", "GET"])
async def micropub(request):
return sanic.response.text("Not Implemented", status=501)
@app.route("/webmention", methods=["POST"])
async def webmention(request):
source = request.form.get('source')
target = request.form.get('target')
# validate urls
if not validators.url(source):
return sanic.response.text('Invalide source url', status=400)
if not validators.url(target):
return sanic.response.text('Invalide target url', status=400)
# check if our site is actually the target for the webmention
_target = urllib.parse.urlparse(target)
if _target.hostname not in shared.config.get('site', 'domains'):
return sanic.response.text('target domain is not me', status=400)
# ignore selfpings
_source = urllib.parse.urlparse(source)
if _source.hostname in shared.config.get('site', 'domains'):
return sanic.response.text('selfpings are not allowed', status=400)
# it is unfortunate that I need to init this every time, but
# otherwise it'll become read-only for reasons I'm yet to grasp
# the actual parsing will be done at site generation time
wdb = shared.WebmentionQueue()
wdb.maybe_queue(source, target)
# telegram notification, if set
l = envelope.Letter(
sender=(
'NASG',
'nasg@%s' % socket.getfqdn()
),
recipient=(
shared.config.get('author', 'name'),
shared.config.get('author', 'email')
),
subject="[webmention] from %s" % _source.hostname,
text='incoming webmention from %s to %s' % (
source,
target
)
)
l.make()
l.send()
response = sanic.response.text("Accepted", status=202)
return response
#app.run(host="127.0.0.1", port=9002, log_config=None)
app.run(host="127.0.0.1", port=9002)
|