TL;DR QueueHandler를 사용해서 process-safe logger를 만들고 LoggerAdapter를 활용해서 매 로그 메시지마다 특정 정보가 자동으로 출력. main.py #!/usr/bin/env python3 import logging import logging.handlers import multiprocessing from worker import worker_process from my_logger import logger_process if __name__ == "__main__": # Configure logging logging.basicConfig(level=logging.DEBUG) # Create a queue for logging log_queue = multiprocessing.Queue() # Create worker processes num_workers = 2 processes = [] for i in range(num_workers): p = multiprocessing.Process(target=worker_process, args=(i, log_queue)) processes.append(p) p.start() # Create a logger process for process-safe loggging p = multiprocessing.
Go, Rust and Python
Python은 20년째 하고 있는 것 같은데 어째 2개월 한 사람보다 못하냐.
http://sosa0sa.com:2368/use-jetpack-for-wordpress-5-0/ 의 연장선. Jetpack plugin의 최신 버전을 알아내서 자동으로 해당 바이너리 파일을 다운로드 받아 보자.
몇 가지 module 을 사용하는데 필요한 module은 https://realpython.com/python-web-scraping-practical-introduction/ 을 참고해서 설치
$ python3 -m venv venv $ . ./venv/bin/activate $ pip3 install requests BeautifulSoup4 webscrap.py from requests import get from requests.exceptions import RequestException from contextlib import closing from bs4 import BeautifulSoup from urllib.request import * ''' https://realpython.com/python-web-scraping-practical-introduction/ ''' def simple_get(url): """ Attempts to get the content at `url` by making an HTTP GET request.
Time-series data를 python을 이용해서 influxDB에 저장하고, Grafana로 그래프를 보여주는 예제
https://github.com/cychong47/influxdb_example.git
Install Grafana and influxDB Install Grafana 직접 호스트에 설치할 수도 있지만, 세상 편하게 만들어준 docker를 이용해서 grafana, influxdb등을 설치하자.
mbpr15:~ cychong$ docker pull grafana/grafana Using default tag: latest latest: Pulling from grafana/grafana f2aa67a397c4: Pull complete 89573effc7c8: Pull complete b55c103da375: Pull complete Digest: sha256:364bec4a39ecbec744ea4270aae35f6554eb6f2047b3ee08f7b5f1134857c32c Status: Downloaded newer image for grafana/grafana:latest Start grafana
mbpr15:~ cychong$ docker run -d -p 3000:3000 —name grafana grafana/grafana 148894d7009259b02b04e1a98467f549400be91f9b055f8686557d69b9339e4b Install influxDB influxdb도 docker 명령어 하나로 설치
It is gone finally… [macOS Server 5.4 changes in High Sierra you need to know about!])https://www.imore.com/changes-macos-server-54-high-sierra)
File Transfer Protocol (FTP): A longtime a security risk, for example for sending password information in clear text, FTP support will be removed from macOS server if you upgrade.
As my xeros printer/scanner suppors only ftp or samba(which is much much slower than the ftp) I have to run my own ftp server in my mac mini.
Just for the record as some environments are mentioned
mbpr15:working cychong$ brew install python ==> Installing dependencies for python: readline, sqlite, gdbm, openssl ==> Installing python dependency: readline ==> Downloading https://homebrew.bintray.com/bottles/readline-7.0.3_1.high_sierra.bottle.tar.gz ######################################################################## 100.0% ==> Pouring readline-7.0.3_1.high_sierra.bottle.tar.gz ==> Caveats This formula is keg-only, which means it was not symlinked into /usr/local, because macOS provides the BSD libedit library, which shadows libreadline. In order to prevent conflicts when programs look for libreadline we are defaulting this GNU Readline installation to keg-only.
정규식으로 http 분석 def processHTTP(data): str_method = "" str_uri = "" 정규표현식을 통해 넘어온 데이터에서 METHOD, URI, HTTP 버전 정보등으로 구분함 h = re.search("(?P<method>(^GET|^POST|^PUT|^DELETE)) (?P<uri>.+) (?P<version>.+)", data) if not h: return "Error" # 정규표현식에 해당하는 데이터가 없는 경우 Error 를 리턴해줌 # method 로 정의된 부준은 str_method 에 저장 if h.group("method"): str_method = h.group("method") # URI 데이터는 str_uri 에 저장 if h.group("uri"): str_uri = h.group("uri") return str_method,str_uri # method 와 uri 를 리턴해 줌 출처 : http://www.
import modules $ python >>> from scapy.all import * >>> from scapy.layers.ipsec import * build plaintext packet >>> p = IP(src='1.1.1.1', dst='2.2.2.2') / TCP(sport=45012, dport=80) / Raw('testdata') >>> p = IP(str(p)) setup SA >>> sa = SecurityAssociation(ESP, spi=0xdeadbeef, crypt_algo='AES-CBC',crypt_key='sixteenbytes key') Encrypt w/o IV >>> e = sa.encrypt(p, 5) >>> e <IP version=4L ihl=5L tos=0x0 len=76 id=1 flags= frag=0L ttl=64 proto=esp chksum=0x747a src=1.1.1.1 dst=2.2.2.2 |<ESP spi=0xdeadbeef seq=5 data='uD\x7fdj19\xe7\xc4\xff8\x10\xcdQ\xf0\xa6\x1e!\x84\xc3>!\x18\xa6\xf6\xb8\x93\xc6it\x9a\xfc\x1c\xee\xe5C\xcd\xf0\x7fD\xca\x8d\xadKh\xa8\xe5x' |>> >>> e.show() ###[ IP ]### version = 4L ihl = 5L tos = 0x0 len = 76 id = 1 flags = frag = 0L ttl = 64 proto = esp chksum = 0x747a src = 1.
다음과 같이 scapy를 이용해서 fragment를 쉽게 만들 수 있다.
from scapy.all import * dip="10.0.0.1" payload=" "*1000 packet=IP(dst=dip)/UDP(dport=0x1234)/payload frag_list=fragment(packet,fragsize=500) counter=1 for fragment in frag_list: print "Packet no%d" %counter print fragment.show() counter+=1 send(fragment) frag_list에서 counter 변수를 확인해서 전송하지 않으면 간단하게 fragment가 수신되지 않은 경우에 시험할 수 있음.
필요하면 frag_list의 순서를 뒤집는 것도 가능하고, 각 fragment의 offset값을 조정하거나 패킷 크기를 변경하면 다른 비정상 경우도 쉽게 시험할 수 있다.
scapy interactive tutorial
Introducing Python 을 판교어린이도서관에서 짧게 보고 적은 아이템들
python3 based
Decorator는 공부가 필요한 내용
sys.path : module 검색 경로 __init__.py 파일이 있으면 그 디렉토리를 PKG로 간주함
defaultdic()
Counters()
dicionary는 key의 순서를 보장하지 않음. OrderedDict()로 사전을 정의하면 가능 deque = stack + queue pprint()는 print보다 깔끔하게 출력한다고.
'\uXXX' 유니코드
%10.4s : 10칸의 공간. 문자열 중 4개만 출력
struct '>LL' : '>' Big endian, L : uint32_t
list comprehension : for loop보다 빠름.