athoune

async def github_webhook(secret :str, request :web.Request) -> dict:
    assert request.content_length < 1000000, "Request content too fat" # 1M
    digest, signature = request.headers['X-HUB-SIGNATURE'].split("=", 1)
    assert digest == "sha1", "Digest must be sha1" # use a whitelist
    body = await request.content.read()
    h = hmac.HMAC(bytes(secret, "UTF8"), msg=body, digestmod=digest)
    assert h.hexdigest() == signature, "Bad signature"
    return json.loads(body.decode('UTF8'))

athoune

#!/usr/bin/env python3

import asyncio


@asyncio.coroutine
def ping(loop, target, dump=False):
    create =  asyncio.create_subprocess_exec('ping', '-c', '10', target,
                                          stdout=asyncio.subprocess.PIPE)
    proc = yield from create

athoune

#!/usr/bin/env python3

import signal
import os

from flask import Flask, abort
from werkzeug.serving import run_simple


state = True

athoune

import os
import sys
import luigi
from raven import Client

client = Client()

# Define your Luigi tasks here

@luigi.Task.event_handler(luigi.Event.FAILURE)

athoune

#!/usr/bin/env python3
from pathlib import Path


def containerd():
	for p in Path("/sys/fs/cgroup/memory/system.slice/containerd.service/").iterdir():
		if not p.is_dir():
			continue
		print(p.name)
		with (p / "memory.stat").open("r") as f:

athoune

#!/usr/bin/env python3

import pwd
import sys
import re
import subprocess
import tempfile
import os
import io
from collections import namedtuple

athoune

#!/usr/bin/env python3
import subprocess
import re

from pypred import Predicate


SPACE = re.compile(r"\s+")

def mailq():

athoune

FROM base/archlinux

RUN pacman-key --populate archlinux &&\
        pacman-key --refresh-keys &&\
        pacman-key -r arojas@us.es

RUN pacman -Sy

RUN yes | pacman -Sy libevent libssh libutempter msgpack-c openssl zlib automake cmake ruby autoconf gcc pkg-config make libunistring git

athoune

#!/usr/bin/env python3

import os
import asyncio
import aiohttp
from aiohttp import web


@asyncio.coroutine
def do_ping(session, domain):

athoune

package pubsub

import (
	"sync"
)

type Pubsub struct {
	channels map[string]map[int]*Channel
	cpt      int
	lock     sync.Mutex