bgpwtf/cccampix/frontend/frontend.py

import logging

import arrow
import grpc
from flask import Flask, render_template
from flask.logging import default_handler

from bgpwtf.cccampix.proto import ix_pb2 as ipb
from bgpwtf.cccampix.proto import ix_pb2_grpc as ipb_grpc

from bgpwtf.cccampix.frontend.channel import Channel


logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(message)s')
logger = logging.getLogger(__name__)


check_info = {
    'irr': ('IRR', 'Required IRR entires are present for this AS'),
    'pgp': ('PGP', 'The PGP key defined in the IRR entry exists on PGP keyservers'),
}


def create_app(config=None):
    app = Flask(__name__)
    app.config.update(config or {})
    app.logger.removeHandler(default_handler)
    verifier = Channel(app, config['verifier'])
    
    
    @app.route('/')
    def view_index():
        req = ipb.ProcessorStatusRequest()
        processors = verifier.stub(ipb_grpc.VerifierStub).ProcessorStatus.future(req)
        req = ipb.PeerSummaryRequest()
        peers = verifier.stub(ipb_grpc.VerifierStub).PeerSummary(req)

        processors = sorted(processors.result().processors, key=lambda el: el.name)
        peers = sorted(list(peers), key=lambda el: el.peeringdb_info.asn)

        return render_template('index.html',
                processors=processors, peers=peers)

    @app.route('/asn/<int:asn>')
    def view_asn(asn):
        req = ipb.PeerDetailsRequest()
        req.asn = asn

        details = None
        try:
            details = verifier.stub(ipb_grpc.VerifierStub).PeerDetails(req)
        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.NOT_FOUND:
                return 'No such ASN.'
            else:
                return 'Internal server error.'
        
        return render_template('asn.html', details=details, asn=asn, check_info=check_info)

    @app.route('/asn/<int:asn>/config.gpg')
    def view_asn_config(asn):
        req = ipb.PeerSecretsRequest()
        req.asn = asn

        details = None
        try:
            details = verifier.stub(ipb_grpc.VerifierStub).PeerSecrets(req)
        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.NOT_FOUND:
                return 'No such ASN.'
            else:
                return 'Internal server error.'

        return details.pgp_data, 200, {'Content-Type': 'application/octet-stream'}
    
    @app.template_filter()
    def from_nano(v):
        return arrow.get(v/1e9)

    return app