[네트워크] Python의 http.server와 CGI

Teus 23.12.07

10,042 29 4

facebook twitter kakao link

29 4

facebook twitter kakao link

DEVOTEE 요약: Python에서는 내장 http.server와 CGI를 활용해 간단한 서버를 생성할 수 있다. http.server는 handler를 선택해 특정 요청에 따른 응답을 줄 수 있으며, CGI는 독립적인 프로세스를 생성해 HTTP 응답 메시지를 만들 수 있다. 그러나 CGI는 요청이 줄어들 때마다 프로세스를 새로 생성하며, 복수의 요청을 병렬로 처리하지는 않는다.

DEVOTEE 추천 블로그

안녕하세요. Teus입니다.

계속해서 네트워크 관련 포스팅으로 찾아뵙고 있습니다.

이번 포스팅은 Python에서 기초적으로 제공하는 http.server와 CGI에 대해서 다룹니다.

1. http.server

Python은 socketserver.TCPServer의 child Class인 http.server.HTTPServer

를 활용해서 기본적인 서버를 만들 수 있게 준비되어있습니다.

import http.server
import socketserver

def run(server_class=http.server.HTTPServer, handler_class=http.server.SimpleHTTPRequestHandler):
    PORT = 8000
    server_address = ('', PORT)
    httpd = server_class(server_address, handler_class)
    httpd.serve_forever()

run()

WSGI의 경우 Server class와 Handler를 선택할 필요 없이 Response를 줄 App만 만들면 되었습니다.

반면 위 http.server를 사용하면 handler를 선택해서 response를 줄 수 있습니다.

이때 http.server.CGIHTTPRequestHanlder를 통해서 CGI가 어떤식으로 동작하는지 알아보도록 하겠습니다.

일단 TCPServer의 serve_forever()같은 경우 WSGI 1편 포스팅 이미 다뤘습니다.

Socket이 selector로 request를 받아들이고

BaseServer의 self.finish_request()에서 ReqeustHandler를 사용해서 처리했습니다.

(이 내용은 WSGI2편 포스팅에서 다뤘습니다.)

#https://github.com/python/cpython/blob/7dead6a33ac786465c3d927d5c2025caf0c1a71b/Lib/socketserver.py#L342
    def process_request(self, request, client_address):
        self.finish_request(request, client_address)
        self.shutdown_request(request)
    ...
        
    def finish_request(self, request, client_address):
        """Finish one request by instantiating RequestHandlerClass."""
        self.RequestHandlerClass(request, client_address, self)

2. CGIHTTPRequestHandler

일단 CGIHTTPRequestHandler는 아래와 같은 구조로 되어있습니다.

BaseRequestHandler

↑

StremRequestHandler

↑

BaseHTTPRequestHandler

↑

SimpleHTTPRequestHandler

↑

CGIHTTPRequestHandler

일단 BaseRequestHandler의 생성자를 다시 보겠습니다.

class BaseRequestHandler:
    def __init__(self, request, client_address, server):
        self.request = request
        self.client_address = client_address
        self.server = server
        self.setup()
        try:
            self.handle()
        finally:
            self.finish()

    def setup(self):
        pass

    def handle(self):
        pass

    def finish(self):
        pass

생성자를 실행하면서 self.setup(), self.handle(), self.finish()를 순서대로 실행하게 됩니다.

이때 self.setup()와 self.finish()은

socket을 준비하고 stdout에 준비된 Data를 socket을 통해서 외부로 송출합니다.

위 method들은 StreamRequestHandler에 정의 되어있습니다.

#https://github.com/python/cpython/blob/7dead6a33ac786465c3d927d5c2025caf0c1a71b/Lib/socketserver.py#L777
class StreamRequestHandler(BaseRequestHandler):
    ...
    rbufsize = -1
    wbufsize = 0

    # A timeout to apply to the request socket, if not None.
    timeout = None

    # Disable nagle algorithm for this socket, if True.
    # Use only when wbufsize != 0, to avoid small packets.
    disable_nagle_algorithm = False

    def setup(self):
        self.connection = self.request
        if self.timeout is not None:
            self.connection.settimeout(self.timeout)
        if self.disable_nagle_algorithm:
            self.connection.setsockopt(socket.IPPROTO_TCP,
                                       socket.TCP_NODELAY, True)
        self.rfile = self.connection.makefile('rb', self.rbufsize)
        if self.wbufsize == 0:
            self.wfile = _SocketWriter(self.connection)
        else:
            self.wfile = self.connection.makefile('wb', self.wbufsize)

    def finish(self):
        if not self.wfile.closed:
            try:
                self.wfile.flush()
            except socket.error:
                # A final socket error may have occurred here, such as
                # the local error ECONNABORTED.
                pass
        self.wfile.close()
        self.rfile.close()

WSGI와 달리, 이번에는 self.handle()이 BaseHTTPRequestHandler에 정의된 상태로 사용됩니다.

#https://github.com/python/cpython/blob/7dead6a33ac786465c3d927d5c2025caf0c1a71b/Lib/http/server.py#L432
class BaseHTTPRequestHandler(socketserver.StreamRequestHandler):
    ...
    def handle(self):
        """Handle multiple requests if necessary."""
        self.close_connection = True

        self.handle_one_request()
        while not self.close_connection:
            self.handle_one_request()
            
    def handle_one_request(self):
        try:
            self.raw_requestline = self.rfile.readline(65537)
            if len(self.raw_requestline) > 65536:
                self.requestline = ''
                self.request_version = ''
                self.command = ''
                self.send_error(HTTPStatus.REQUEST_URI_TOO_LONG)
                return
            if not self.raw_requestline:
                self.close_connection = True
                return
            if not self.parse_request():
                # An error code has been sent, just exit
                return
            mname = 'do_' + self.command
            if not hasattr(self, mname):
                self.send_error(
                    HTTPStatus.NOT_IMPLEMENTED,
                    "Unsupported method (%r)" % self.command)
                return
            #여기서 하부의 do_GET, do_POST와 같은 method를 가지고옴
            method = getattr(self, mname)
            method()
            self.wfile.flush() #actually send the response if not already done.
        ...

위 self.handle_one_request()에서 볼 수 있듯 do_ResponseMethod를 통해서 HTTP Msg를 처리하게 됩니다.

do_GET,do_GET은 SimpleHTTPRequestHandler

do_POST는 CGIHTTPRequestHandler에 정의되어 있습니다.

그래서 SimpleHTTPRequestHandler를 사용한 HTTP Server는 Post를 사용할 경우 Error가 발생합니다

이때 do_POST를 통해서 CGI가 어떻게 Python에서 구현되었는지 볼까요?

#https://github.com/python/cpython/blob/7dead6a33ac786465c3d927d5c2025caf0c1a71b/Lib/http/server.py#L978
class CGIHTTPRequestHandler(SimpleHTTPRequestHandler):
    ...
    def do_POST(self):
    #cgi가 가능한 url path인지를 판별
    if self.is_cgi():
        self.run_cgi()
    else:
       #Error send
       
    def is_cgi(self):        
        collapsed_path = _url_collapse_path(self.path)
        dir_sep = collapsed_path.find('/', 1)
        while dir_sep > 0 and not collapsed_path[:dir_sep] in self.cgi_directories:
            dir_sep = collapsed_path.find('/', dir_sep+1)
        if dir_sep > 0:
            head, tail = collapsed_path[:dir_sep], collapsed_path[dir_sep+1:]
            self.cgi_info = head, tail
            return True
        return False
    cgi_directories = ['/cgi-bin', '/htbin']

CGI의 경우 response를 보낼 app을 /cgi-bin 혹은 /htbin 위치에 두어야만 작동합니다

이를 python코드로 cgi_directories를 만들어놓고, 여기서 path가 이중에 있는지를 찾는걸 볼 수 있습니다.

그리고, 문제가 없다면 self.run_cgi를 통해서 CGI Script를 실행합니다.

#https://github.com/python/cpython/blob/7dead6a33ac786465c3d927d5c2025caf0c1a71b/Lib/http/server.py#L1053
    def run_cgi(self):
        """Execute a CGI script."""
        #==============================================
        #dir = '/cgi-bin' or '/htbin'
        #rest = cgi script file.py
        #==============================================
        dir, rest = self.cgi_info
        path = dir + '/' + rest
        i = path.find('/', len(dir)+1)
        while i >= 0:
            nextdir = path[:i]
            nextrest = path[i+1:]

            scriptdir = self.translate_path(nextdir)
            if os.path.isdir(scriptdir):
                dir, rest = nextdir, nextrest
                i = path.find('/', len(dir)+1)
            else:
                break

        # find an explicit query string, if present.
        rest, _, query = rest.partition('?')

        # dissect the part after the directory name into a script name &
        # a possible additional path, to be stored in PATH_INFO.
        i = rest.find('/')
        if i >= 0:
            script, rest = rest[:i], rest[i:]
        else:
            script, rest = rest, ''

        scriptname = dir + '/' + script
        scriptfile = self.translate_path(scriptname)
        if not os.path.exists(scriptfile):
            self.send_error(
                HTTPStatus.NOT_FOUND,
                "No such CGI script (%r)" % scriptname)
            return
        if not os.path.isfile(scriptfile):
            self.send_error(
                HTTPStatus.FORBIDDEN,
                "CGI script is not a plain file (%r)" % scriptname)
            return
        ispy = self.is_python(scriptname)
        if self.have_fork or not ispy:
            if not self.is_executable(scriptfile):
                self.send_error(
                    HTTPStatus.FORBIDDEN,
                    "CGI script is not executable (%r)" % scriptname)
                return
        #==============================================
        # 여기까지 CGI Script를 찾고, CGI Sciprt가 유효한지를 판단함
        #==============================================

먼저 CGI Script 실행하기 앞서

주어진 url 주소를 파싱해서 CGI Script의 정확한 위치를 찾고

해당 Script가 실제 존재하는지, python으로 실행 가능한 파일인지를 검증합니다.

    def run_cgi(self):
        ...
        env = copy.deepcopy(os.environ)
        env['SERVER_SOFTWARE'] = self.version_string()
        env['SERVER_NAME'] = self.server.server_name
        env['GATEWAY_INTERFACE'] = 'CGI/1.1'
        env['SERVER_PROTOCOL'] = self.protocol_version
        env['SERVER_PORT'] = str(self.server.server_port)
        env['REQUEST_METHOD'] = self.command
        uqrest = urllib.parse.unquote(rest)
        env['PATH_INFO'] = uqrest
        env['PATH_TRANSLATED'] = self.translate_path(uqrest)
        env['SCRIPT_NAME'] = scriptname
        env['QUERY_STRING'] = query
        env['REMOTE_ADDR'] = self.client_address[0]
        authorization = self.headers.get("authorization")
        if authorization:
            authorization = authorization.split()
            if len(authorization) == 2:
                import base64, binascii
                env['AUTH_TYPE'] = authorization[0]
                if authorization[0].lower() == "basic":
                    try:
                        authorization = authorization[1].encode('ascii')
                        authorization = base64.decodebytes(authorization).\
                                        decode('ascii')
                    except (binascii.Error, UnicodeError):
                        pass
                    else:
                        authorization = authorization.split(':')
                        if len(authorization) == 2:
                            env['REMOTE_USER'] = authorization[0]
        # XXX REMOTE_IDENT
        if self.headers.get('content-type') is None:
            env['CONTENT_TYPE'] = self.headers.get_content_type()
        else:
            env['CONTENT_TYPE'] = self.headers['content-type']
        length = self.headers.get('content-length')
        if length:
            env['CONTENT_LENGTH'] = length
        referer = self.headers.get('referer')
        if referer:
            env['HTTP_REFERER'] = referer
        accept = self.headers.get_all('accept', ())
        env['HTTP_ACCEPT'] = ','.join(accept)
        ua = self.headers.get('user-agent')
        if ua:
            env['HTTP_USER_AGENT'] = ua
        co = filter(None, self.headers.get_all('cookie', []))
        cookie_str = ', '.join(co)
        if cookie_str:
            env['HTTP_COOKIE'] = cookie_str
        # XXX Other HTTP_* headers
        # Since we're setting the env in the parent, provide empty
        # values to override previously set values
        for k in ('QUERY_STRING', 'REMOTE_HOST', 'CONTENT_LENGTH',
                  'HTTP_USER_AGENT', 'HTTP_COOKIE', 'HTTP_REFERER'):
            env.setdefault(k, "")

        self.send_response(HTTPStatus.OK, "Script output follows")
        #header를 response buff에 입력한 뒤
        #header를 담고있던 list를 초기화 시켜줌
        self.flush_headers()

        decoded_query = query.replace('+', ' ')

그리고 header를 준비합니다.

header같은 경우 dict형태로 저장하는것을 볼 수 있습니다.

여기서 self.send_response()라는 method이름 때문에 혼란스러울 수 있으나

사실 self.send_response()는 client한태 보내는것이 아니라, buffer에다가 보내는 것을 확인할 수 있습니다.

#https://github.com/python/cpython/blob/7dead6a33ac786465c3d927d5c2025caf0c1a71b/Lib/http/server.py#L520
class BaseHTTPRequestHandler(socketserver.StreamRequestHandler):
    ...
    def send_response(self, code, message=None):
        """Add the response header to the headers buffer and log the
        response code.

        Also send two standard headers with the server software
        version and the current date.
        """
        self.log_request(code)
        self.send_response_only(code, message)
        self.send_header('Server', self.version_string())
        self.send_header('Date', self.date_time_string())

    def send_response_only(self, code, message=None):
        """Send the response header only."""
        if self.request_version != 'HTTP/0.9':
            if message is None:
                if code in self.responses:
                    message = self.responses[code][0]
                else:
                    message = ''
            if not hasattr(self, '_headers_buffer'):
                self._headers_buffer = []
            self._headers_buffer.append(("%s %d %s\r\n" %
                    (self.protocol_version, code, message)).encode(
                        'latin-1', 'strict'))

이제 HTTP Header와 Script가 준비과 완료 되었습니다.

이제 이 Script를 실행시켜 HTTP Response Msg를 준비할 시간입니다.

    #실제 CGI를 실행
    def run_cgi(self):
        ...
        if self.have_fork:
            # Unix -- fork as we should
            # Unix 계열 시스템에서 fork를 사용한 CGI실행
            ...
        else:
            # Non-Unix -- use subprocess
            import subprocess
            cmdline = [scriptfile]
            if self.is_python(scriptfile):
                interp = sys.executable
                if interp.lower().endswith("w.exe"):
                    # On Windows, use python.exe, not pythonw.exe
                    interp = interp[:-5] + interp[-4:]
                cmdline = [interp, '-u'] + cmdline
            if '=' not in query:
                cmdline.append(query)
            #==============================================
            #여기까지 Command prompt에 python파일을 실행하기위한
            #실행문장 + 매개변수를 정의함
            #==============================================        
            self.log_message("command: %s", subprocess.list2cmdline(cmdline))
            try:
                nbytes = int(length)
            except (TypeError, ValueError):
                nbytes = 0
            #subprocess를 통해서 response를 할 때 마다 Process를 새로 만듦
            p = subprocess.Popen(cmdline,
                                 stdin=subprocess.PIPE,
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE,
                                 env = env
                                 )
            if self.command.lower() == "post" and nbytes > 0:
                data = self.rfile.read(nbytes)
            else:
                data = None
            # throw away additional data [see bug #427345]
            while select.select([self.rfile._sock], [], [], 0)[0]:
                if not self.rfile._sock.recv(1):
                    break
            # 자식 process에서 스크립트를 실행하고
            # 스크립트가 끝날 때까지 대기함
            # process.join()과 유사함
            stdout, stderr = p.communicate(data)
            self.wfile.write(stdout)
            if stderr:
                self.log_error('%s', stderr)
            p.stderr.close()
            p.stdout.close()
            status = p.returncode
            if status:
                self.log_error("CGI script exit status %#x", status)
            else:
                self.log_message("CGI script exited OK")```

결과로 확인할 수 있듯

Process를 만들어서 HTTP Response Msg를 만들어 냅니다.

그리고 Main Process에서는 subprocess에서 만들어낸 text를 기반으로

HTTP Response Msg를 만들고, Socket을 통해 외부로 전달합니다.

의외인 점은, HTTP Msg를 보내는 작업 전체가 subprocess로 되어있지 않습니다.

때문에 N개의 Request가 왔을 때 N개의 Process가 생성되어 동시에 처리할 수 있는게 아니라

N개의 Request를 순차적으로 처리하게 됩니다.

3. CGI 서버 만들어보기

먼저 project directory를 아래처럼 만들어 줍니다.

simple_server(폴더)
->server.py
->cgi-bin(폴더)
-->cgi_script.py

그리고 server.py 파일을 아래처럼 http server로 구축 합니다.

#simple_server/server.py
import http.server
import socketserver

PORT = 8000

Handler = http.server.CGIHTTPRequestHandler

with http.server.HTTPServer(("", PORT), Handler) as httpd:
    print("serving at port", PORT)
    httpd.serve_forever()

그리고 response를 만들기 위한 script파일을 아래처럼 만들어 줍니다.

#simple_server/cgi-bin/cgi_script.py
import time

time.sleep(3)
print("Content-type: text/html\n\n")
print("hello world!")

그리고 서버를 실행하면 serving at port 8000 이라는 메세지와 함께 서버가 작동합니다.

cgi는 post로만 작동하기 때문에 python requests package로 실행 해봅니다.

import requests
res = requests.post("http://localhost:8000/cgi-bin/cgi_script.py")
print(res.text)

실행하면 서버shell에서 정상적으로 script가 실행되고 response를 보낸것을 확인할 수 있습니다.

이때 두개의 python shell을 키고 동시에 실행할 경우 순차적으로 두개가 실행되는것을 통해서

CGI가 동시에 여러 requests를 처리하지는 않는것을 볼 수 있습니다.

#Python #Programming Language #Flask

등록

DEVOTEE를 활성화 시키면
지금 작성한 댓글에 AI가 댓글을 달아줍니다.

익명

Teus 님의 최신 블로그

LLM의 thinking_token_budget은 어떤 역할을 하는가?

안녕하세요. teus입니다. 이번 포스팅은, opencode 등의 LLM에서 설정할 수 있는 thinking_token_budget에 대해서 다룹니다. 0. Effort opencode나 claude code의 설정에 보면, 모델의 reasoningEffort를 수정 할 수 있습니다. https://opencode.ai/docs/models/#configure-models 문제는 위와같이 reasoningEffort라고 지정되어있는 변수의 경우 직접적으로 어느정도의 reasoning을 진행할 지를 우리가 알 수는 없고 enable_thinking 활성화 여부와 system prompt의 일부분에 영향을 끼친다고 이야기 했습니다. claude code의 reason...
claude code의 reasoning effort는 vllm에도 효과가 있는가?

안녕하세요. teus입니다. 이번 포스팅은, claude code에서 설정할 수 있는 reasoning effort가 과연 효과가 있는지 소스코드 레벨에서 점검해봅니다. 0. Effort claude code를 보면, 위처럼 effort를 설정할 수 있습니다. 원래 claude 모델을 쓸 때는 위와같은 설정을 쓰는게 가정하고 학습이 되었겠지만 사내에서는 claude 모델이 아니라, vLLM기반의 OSS모델을 사용 중 입니다. 그러면 과연 해당 기능이 제대로 동작할 수 있을까요? GLM 5.2 모델 기준 flow(정리본) Claude Code /effort high │ ▼ POST /v1/messages output_config.effort = ＂high＂ │ ▼ ...
AI Agent 속도 최적화를 위한 Speculative Decoding

안녕하세요. Teus입니다. 이번 포스팅은 Speculative Decoding이라는 방법을 통해서 어떻게하면 AI Agent의 Latency를 개선할 수 있을지 생각해 봅니다. 1. Speculative Decoding? Speculative는 구글에서 검색해보면 Speculative : 투기적인 이라는 단어로 검색됩니다. Speculative Decoding은 CPU의 분기예측과 비슷한데 앞으로 해야 할 행동을 몇가지 찾아놓고 그 행동이 맞으면 현실로 옮기고, 아닐경우 폐기하는 컨셉을 통해서 AI Agent의 Token 속도를 올리는 방법입니다. 2. 어떻게? 원리를 비유적으로 표현하면 학생이 빠르게 대충 문제를 풀게하고 선생님 다수가 그 결과를 체점해서 틀린...
A2A 튜토리얼 코드 톺아보기

안녕하세요. Teus입니다. 이번 포스팅은 A2A(Agent 2 Agent) Protocol이 뭔지 간단히 알아보고 부실한 tutorial을 채우기 위한 공식 튜토리얼 코드를 간단히 톺아보겠습니다. 0. A2A A2A(Agent To Agent)는 Google에서 25년 4월 9일 구글에서 발표한 프로토콜 입니다. 이름에서 볼 수 있듯, AI Agent 〈-〉 AI Agent간 통신을 위한 프로토콜 입니다. MCP를 사용 할때까지는 MCP를 통해서 MCP Server에 저장된 Tools, Resource, Prompt를 불러오고 이 Tools를 활용해서 Agent를 각자 구축 하는 방식 이었습니다. 이제 A2A를 이용해서 Agent를 각자 구현하지 말고 각자 만든...

DEVOTEE 추천 블로그

[네트워크] WSGI와 Python 1편

안녕하세요 Teus 입니다. 오랜만에 RN 스터디가 아닌 글로 찾아왔습니다. 이번시간에 다뤄볼 내용은 WGSI입니다. WSGI에 앞서, 모든 프로그래밍언어에서 사용되는 CGI에 대해서 간단하게 알아봅니다. 0. CGI는 무엇인가? CGI는 Common Gate Interface로 Client로 부터 발송된 Request HTTP Msg를 가장 먼저 받아들이는 Layer입니다. Interface이기 때문에, 특정 언어에 귀속된게 아니라 다양한 프로그래밍 언어에서 구현이 되어 있습니다. 과거 Web 프로그래밍이 난잡하던 시기에, 모든 언어에서 동일한 Interface로 HTTP Msg를 처리할 수 있게 만들기위해서 고안된 알고 있습니다. CGI의 특징으로는 HTTP ...
[네트워크] WSGI와 Python 2편

안녕하세요. Teus입니다. 이번 포스팅은 지난시간에 다뤘던 Python WSGI 2탄 입니다. 지난 포스팅에서 활용했던 WSGI를 사용하는 기초 예제를 기반으로 설명을 진행 합니다. 1. WSGI의 Response처리 WSGI는 기본적으로 아래 두가지로 구성됩니다. 무한루프를 돌면서 Request를 받아들이는 socket 서버 socket서버가 받아들인 request를 처리하는 requesthandler 이번 포스팅에서는 app과 requesthanlder를 활용해서 response를 보내는 방법을 알아 볼겁니다. 대략적으로 response과정에서 app을 callback으로 활용해서 request를 처리할 것을 예상은 할 수 있습니다. 실제로 그런지 알아 볼까...
[C/PYTHON] Socket 기초 1 - 간단한 http 서버 만들기

안녕하세요. SKHYNIX DRAM노광PHOTO기술팀 장우석 TL입니다. socket은 운영체제에 내장된, 전화를 거는 기능입니다. 조금 더 고급지게 말하면, socket은 응용 계층(L4)에서 그 아래 계층(L3~L1)에 접근하기 위한 인터페이스입니다. HTTP 서버를 구성할 때 사용하는 DJANGO, SPRING 등의 라이브러리들도 socket을 통해 네트워크에 접근합니다. 이 글에서는 socket을 이용하여 http 서버를 만드는 코드를 c언어와 python으로 보여드리고 코드를 설명 드리겠습니다. 1. simple http server for linux //server.c //simple http server for linux //gcc server.c -...

목록으로

DEVOCEAN

로그인 선택

미리보기

커뮤니티

글 등록

임시저장함

데보션 블로그 게재 요청