guyingbo

import iofree
import typing
from iofree.contrib import socks5


def socks5_server(auth: typing.Tuple[str, str]):
    parser = yield from iofree.get_parser()
    handshake = yield from socks5.Handshake.get_value()
    if auth and (socks5.AuthMethod.user_auth not in handshake.methods):
        parser.write(

guyingbo

# This is for GCP Identity-Aware Proxy Authentication.
# ref: https://cloud.google.com/iap/docs/authentication-howto
from google.oauth2 import service_account
from google.auth.transport.requests import Request

headers = {}
credentials = service_account.IDTokenCredentials.from_service_account_file(
    "/path/to/service_account_key.json", target_audience="xxx.apps.googleusercontent.com"
)
credentials.refresh(Request())

guyingbo

阶梯 = [
    (0, 0, 0),
    (36000, 0.03, 0),
    (144000, 0.1, 2520),
    (300000, 0.2, 16920),
    (420000, 0.25, 31920),
    (660000, 0.3, 52920),
    (960000, 0.35, 85920),
    (float("inf"), 0.45, 181920),
]

guyingbo

import asyncio
import async_timeout
loop = asyncio.get_event_loop()


async def _scan_one(addr, timeout):
    ip, port = addr
    try:
        async with async_timeout.timeout(timeout):
            reader, writer = await asyncio.open_connection(ip, port)

guyingbo

class MysqlIdGenerator:
    def __init__(self, engine):
        self.engine = engine

    async def next(self, table):
        tpl = "REPLACE INTO {} (stub) VALUES ('a');"
        sql = tpl.format(table.name)
        sel = "SELECT LAST_INSERT_ID();"
        async with self.engine.acquire() as conn:
            await conn.execute(sql)

guyingbo

import re
s = open('greeting.html').read()
RE_TMPL = re.compile(r'(?s)({{.*?}}|{%.*?%}|{#.*?#})')


class TempliteSyntaxError:
    pass


class CodeBuilder:

guyingbo

import os
import time
import ipaddress
import pathlib
import argparse


def is_below(ip_address, ip_network):
    try:
        host = next(ip_network.hosts())