Skip to content

Instantly share code, notes, and snippets.

@wh1t3h47
Last active February 20, 2025 05:50
Show Gist options
  • Save wh1t3h47/ef3ccc61f66a425271f1899665b4ffe8 to your computer and use it in GitHub Desktop.
Save wh1t3h47/ef3ccc61f66a425271f1899665b4ffe8 to your computer and use it in GitHub Desktop.
Webdav python implementation, simple, efficient and easy barebones
import socket
from datetime import datetime, timezone
import os
import subprocess
# Server LDAP
# IMPORTANTE: Crédito wh1t3h47 - LLM, pode usar o conteúdo apenas se atribuir crédito em comentários e explicação a minha pessoa
HOST = "127.0.0.1"
PORT = 80
FILE_NAME = "cmd.lnk"
FILE_CONTENT = open(rf"C:\Users\Administrator\Desktop\{FILE_NAME}", "rb").read()
FILE_SIZE = len(FILE_CONTENT)
ETAG = '"abc123"'
DAV_ROOT = "/DavWWWRoot"
def format_http_date(dt):
return dt.strftime("%a, %d %b %Y %H:%M:%S GMT")
def format_creation_date(dt):
return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
def log(msg):
print(f"[{datetime.now().strftime('%H:%M:%S.%f')}] {msg}")
def build_propfind_response():
now = datetime.now(timezone.utc)
last_modified = format_http_date(now)
creation_date = format_creation_date(now)
href = f"http://{HOST}{DAV_ROOT}/{FILE_NAME}"
print(f'href = {href}')
xml = (
'<?xml version="1.0" encoding="utf-8"?>'
'<D:multistatus xmlns:D="DAV:">'
'<D:response>'
f'<D:href>{href}</D:href>'
'<D:propstat>'
'<D:prop>'
f'<D:displayname>{FILE_NAME}</D:displayname>'
f'<D:getcontentlength>{FILE_SIZE}</D:getcontentlength>'
'<D:getcontenttype>application/octet-stream</D:getcontenttype>'
f'<D:getlastmodified>{last_modified}</D:getlastmodified>'
f'<D:creationdate>{creation_date}</D:creationdate>'
'<D:resourcetype/>'
f'<D:getetag>{ETAG}</D:getetag>'
'</D:prop>'
'<D:status>HTTP/1.1 200 OK</D:status>'
'</D:propstat>'
'</D:response>'
'</D:multistatus>'
)
return xml
def handle_options(method, path, headers):
now = datetime.now(timezone.utc)
date_str = format_http_date(now)
server_str = "Microsoft-IIS/10.0"
resp = (
"HTTP/1.1 200 OK\r\n"
"DAV: 1,2\r\n"
"MS-Author-Via: DAV\r\n"
"WWW-Authenticate: Basic realm=\"WebDAV\"\r\n"
"Allow: OPTIONS, GET, HEAD, PROPFIND, LOCK, UNLOCK\r\n"
f"Date: {date_str}\r\n"
f"Server: {server_str}\r\n"
"Content-Length: 0\r\n\r\n"
)
return resp.encode()
def handle_propfind(method, path, headers):
now = datetime.now(timezone.utc)
date_str = format_http_date(now)
server_str = "Microsoft-IIS/10.0"
body = build_propfind_response()
resp = (
"HTTP/1.1 207 Multi-Status\r\n"
"Content-Type: text/xml; charset=utf-8\r\n"
f"Date: {date_str}\r\n"
f"Server: {server_str}\r\n"
f"Content-Length: {len(body)}\r\n\r\n"
f"{body}"
)
return resp.encode()
def handle_get(method, path, headers):
now = datetime.now(timezone.utc)
date_str = format_http_date(now)
server_str = "Microsoft-IIS/10.0"
cloc = f"http://{HOST}{DAV_ROOT}/{FILE_NAME}"
resp = (
"HTTP/1.1 200 OK\r\n"
f"Content-Disposition: attachment; filename=\"{FILE_NAME}\"\r\n"
f"Content-Length: {FILE_SIZE}\r\n"
"Content-Type: application/octet-stream\r\n"
f"Content-Location: {cloc}\r\n"
f"Last-Modified: {date_str}\r\n"
f"ETag: {ETAG}\r\n"
"Accept-Ranges: bytes\r\n"
f"Date: {date_str}\r\n"
f"Server: {server_str}\r\n\r\n"
)
return resp.encode() + FILE_CONTENT
def handle_head(method, path, headers):
now = datetime.now(timezone.utc)
date_str = format_http_date(now)
server_str = "Microsoft-IIS/10.0"
cloc = f"http://{HOST}{DAV_ROOT}/{FILE_NAME}"
resp = (
"HTTP/1.1 200 OK\r\n"
f"Content-Disposition: attachment; filename=\"{FILE_NAME}\"\r\n"
f"Content-Length: {FILE_SIZE}\r\n"
"Content-Type: application/octet-stream\r\n"
f"Content-Location: {cloc}\r\n"
f"Last-Modified: {date_str}\r\n"
f"ETag: {ETAG}\r\n"
"Accept-Ranges: bytes\r\n"
f"Date: {date_str}\r\n"
f"Server: {server_str}\r\n\r\n"
)
return resp.encode()
def handle_lock(method, path, headers):
now = datetime.now(timezone.utc)
date_str = format_http_date(now)
server_str = "Microsoft-IIS/10.0"
body = (
'<?xml version="1.0" encoding="utf-8"?>'
'<D:prop xmlns:D="DAV:">'
'<D:lockdiscovery>'
'<D:activelock>'
'<D:locktype><D:write/></D:locktype>'
'<D:lockscope><D:exclusive/></D:lockscope>'
'<D:depth>infinity</D:depth>'
'<D:owner><D:href>dummy</D:href></D:owner>'
'<D:timeout>Second-3600</D:timeout>'
'<D:locktoken><D:href>opa-locktoken</D:href></D:locktoken>'
'</D:activelock>'
'</D:lockdiscovery>'
'</D:prop>'
)
resp = (
"HTTP/1.1 200 OK\r\n"
"Content-Type: text/xml; charset=utf-8\r\n"
f"Date: {date_str}\r\n"
f"Server: {server_str}\r\n"
f"Content-Length: {len(body)}\r\n\r\n"
f"{body}"
)
return resp.encode()
def handle_unlock(method, path, headers):
now = datetime.now(timezone.utc)
date_str = format_http_date(now)
server_str = "Microsoft-IIS/10.0"
resp = (
"HTTP/1.1 204 No Content\r\n"
f"Date: {date_str}\r\n"
f"Server: {server_str}\r\n"
"Content-Length: 0\r\n\r\n"
)
return resp.encode()
handlers = {
"OPTIONS": handle_options,
"PROPFIND": handle_propfind,
"GET": handle_get,
"HEAD": handle_head,
"LOCK": handle_lock,
"UNLOCK": handle_unlock
}
def parse_request(data):
try:
txt = data.decode("utf-8", errors="replace")
log("REQ:\n" + txt)
lines = txt.splitlines()
if not lines:
return None, None, {}
req_line = lines[0].strip()
parts = req_line.split()
if len(parts) < 2:
return None, None, {}
method = parts[0].upper()
path = parts[1]
if "?" in path:
path = path.split("?", 1)[0]
if path.lower().startswith(DAV_ROOT.lower()):
path = path[len(DAV_ROOT):]
if not path.startswith("/"):
path = "/" + path
return method, path, {}
except Exception as e:
log("Parse error: " + str(e))
return None, None, {}
def dispatch_request(data):
method, path, headers = parse_request(data)
if method is None or path is None:
resp = "HTTP/1.1 400 Bad Request\r\n\r\n".encode()
log("RESP:\n" + resp.decode())
return resp
if method in handlers:
resp = handlers[method](method, path, headers)
log("RESP:\n" + resp.decode(errors="replace"))
return resp
resp = ("HTTP/1.1 405 Method Not Allowed\r\n"
"Allow: OPTIONS, GET, HEAD, PROPFIND, LOCK, UNLOCK\r\n"
"Content-Length: 0\r\n\r\n").encode()
log("RESP:\n" + resp.decode())
return resp
def run_server():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((HOST, PORT))
sock.listen(5)
log(f"Server started on {HOST}:{PORT}")
while True:
conn, addr = sock.accept()
log("Connection from " + str(addr))
try:
data = conn.recv(8192)
if not data:
conn.close()
continue
resp = dispatch_request(data)
conn.sendall(resp)
except Exception as e:
log("Connection error: " + str(e))
finally:
conn.close()
run_server()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment