Python在网络安全领域应用广泛,常见的有PoC编写、爬虫以及木马编写等,在这里将会列举常用的部分
Socket编程是众多C/S架构程序基础,比如游戏、Web服务器以及多数的木马都是基于Socket来实现的,在这里会涉及到TCP/IP协议,但是在这里默认大家都会,下面给出一个OSI等相关模型的概念图
Socket(套接字)是一种编程接口,一般面向网络层和传输层协议(套接字并不限定于TCP/IP),每个套接字绑定一个ip和一个端口上,不同的端口对应不同的服务,比如Web大多以80/443作为对外绑定的端口
Python提供类两个基本的socket模块,其中socket它提供了标准的BSD Socket API,另一个socketserver为服务端编程提供了进一步封装,可以简化网络服务器的开发,函数声明如下所示
socket(family, type[,protocal])其中,family表示套接字对象使用的地址族,可选值有AF_INET(IPv4地址族)、AF_INET6(IPv6地址族)以及AF_UNIX(针对类UNIX系统的套接字);type为可使用的类型,具体可使用的如下所示
socket.SOCK_STREAM # 基于TCP的流式socket通信
socket.SOCK_DGRAM # 基于UDP的数据报式socket通信
socket.SOCK_RAW # 原始套接字(ICMP、IGMP等),且可以通过IP_HDRINCL构造IP头
socket.SOCK_SEQPACKET # 可靠的连续数据包服务第三个参数protocal则是协议类型,默认是0表示套接字,不需要关心该参数,创建TCP Socket的方法如下
sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM)创建UDP Socket的方法如下所示
sock = socket.socket(socket.AF_INET,socket.SOCK_DGRAN)对于Socket编程中的客户端部分示例代码如下所示
import socket, sys
class Client:
def __init__(self, host):
self.host = host
def connet(self):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
print("Socket 创建成功")
except socket.error as err:
print(f"由于 {err} 导致 Socket 创建失败")
sys.exit(1)
try:
remote_ip = socket.gethostbyname(self.host) # 获取远程主机ip
except socket.gaierror:
print("主机名解析失败")
sys.exit(1)
try:
s.connect((remote_ip,80)) # 连接远程主机
print(f"连接到 {self.host} 的远程主机成功")
message = b'GET / HTTP/1.1\r\n\r\n' # 发送数据(\r\n\r\n是http协议要求)
s.sendall(message)
print("数据发送成功")
reply = s.recv(4096) # 接收数据
print("数据接收成功")
print(reply.decode())
s.close()
except socket.error as err:
print(f"连接到远程主机失败: {err}")
sys.exit(1)
if __name__ == "__main__":
cl = Client("www.baidu.com")
cl.connet()最终的输出结果如下所示
数据接收成功
HTTP/1.1 200 OK
Accept-Ranges: bytes
Cache-Control: no-cache
Connection: keep-alive
Content-Length: 29506
Content-Type: text/html
Date: Sat, 10 Jan 2026 10:46:57 GMT
P3p: CP=" OTI DSP COR IVA OUR IND COM "
P3p: CP=" OTI DSP COR IVA OUR IND COM "
Pragma: no-cache
Server: BWS/1.1
Set-Cookie: BAIDUID=DA9892C61FD4E7E2793050D0FCB3D80F:FG=1; expires=Thu, 31-Dec-37 23:55:55 GMT; max-age=2147483647; path=/; domain=.baidu.com
Set-Cookie: BIDUPSID=DA9892C61FD4E7E2793050D0FCB3D80F; expires=Thu, 31-Dec-37 23:55:55 GMT; max-age=2147483647; path=/; domain=.baidu.com
Set-Cookie: PSTM=1768042017; expires=Thu, 31-Dec-37 23:55:55 GMT; max-age=2147483647; path=/; domain=.baidu.com
Set-Cookie: BAIDUID=DA9892C61FD4E7E200252E988C8D6B3E:FG=1; max-age=31536000; expires=Sun, 10-Jan-27 10:46:57 GMT; domain=.baidu.com; path=/; version=1; comment=bd
Tr_id: pr_0xd7a5b47b00574fcb
Traceid: 176804201717823713389505042989565703200
Vary: Accept-Encoding
X-Ua-Compatible: IE=Edge,chrome=1
X-Xss-Protection: 1;mode=block
<!DOCTYPE html>
<html>
<head>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
<meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1" />
<meta content="always" name="referrer" />
<meta
name="description"
content="全球领先的中文搜索引擎、致力于让网民更便捷地获取信息,找到所求。百度超过千亿的中文网页数据库,可以瞬间找到相关的搜索结果。"
/>
<link rel="shortcut icon" href="//www.baidu.com/favicon.ico" type="image/x-icon" />
<link
rel="search"
type="application/opensearchdescription+xml"
href="//www.baidu.com/content-search.xml"
title="百度搜索"
/>
<title>百度一下,你就知道</title>
<style type="text/css">
body {
margin: 0;
padding: 0;
text-align: center;
background: #fff;
height: 100%;
}
html {
overflow-y: auto;
color: #000;
overflow: -moz-scrollbars;
height: 100%;
}
body, input {
font-size: 12px;
font-family: "PingFang SC", Arial, "Microsoft YaHei", sans-serif;
}
a {
text-decoration: none;
}
a:hover {
text-decoration: underline;
}
img {
border: 0;
-ms-interpolation-mode: bicubic;
}
input {
font-size: 100%;
border: 0;
}
body, form {
position: relative;
z-index: 0;
}
#wrapper {
height: 100%;
}
#head .s-ps-islite {
_padding-bottom: 370px;
}
#head_wrapper.s-ps-islite {
padding-bottom: 370px;
}
#head_wrapper.s-ps-islite .s_form {
position: relative;
z-index: 1;
}
#head_wrapper.s-ps-islite .fm {
position: absolute;
bottom: 0;
}
#head_wrapper.s-ps-islite .s-p-top {
position: absolute;
bottom: 40px;
width: 100%;
height: 181px;
}
#head_wrapper.s-ps-islite #s_lg_img {
position: static;
margin: 33px auto 0 auto;
left: 50%;
}
#form {
z-index: 1;
}
.s_form_wrapper {
height: 100%;
}
#lh {
margin: 16px 0 5px;
word-spacing: 3px;
}
.c-font-normal {
font: 13px/23px Arial, sans-serif;
}
.c-color-t {
color: #222;
}
.c-btn,
.c-btn:visited {
color: #333 !important;
}
.c-btn {
display: inline-block;
overflow: hidden;
font-family: inherit;
font-weight: 400;
text-align: center;
vertical-align: middle;
outline: 0;
border: 0;
对Socket客户端编程的主要步骤为:
1.创建套接字
2.连接服务器
3.发送数据
4.接收数据
5.关闭连接
服务端主要是做监听,因此和客户端是有一定的区别,示例代码如下所示
import socket, sys
class Server:
def __init__(self,ip,port):
self.ip = ip
self.port = port
def start(self):
s = socket.socket(socket.AF_INET,socket.SOCK_STREAM) # 创建TCP套接字
try:
s.bind((self.ip,self.port)) # 绑定IP和端口
s.listen(5) # 监听连接
print("等待客户端连接...")
conn,addr = s.accept() # 接受连接
print(f"客户端已连接: {addr[0]}:{addr[1]}") # 打印客户端地址
data = conn.recv(1024) # 接收数据
print(f"收到数据: {data.decode('utf-8')}") # 打印
conn.sendall(bytes("你好, 客户端\n\r", encoding='utf-8')) # 发送数据
conn.close() # 关闭连接
except socket.error as e:
print(f"Socket错误: {e}")
sys.exit(1)
finally:
s.close() # 关闭套接字
if __name__ == "__main__":
server = Server("127.0.0.1", 8080)
server.start()在客户端进行如下操作后会收到相关信息
nc 127.0.0.1 8080
你好,服务端,我是客户端
你好, 客户端
之后在服务端可以看到相关信息
等待客户端连接...
客户端已连接: 127.0.0.1:57452
收到数据: 你好,服务端,我是客户端
在网络安全领域中,很多时候对于检测到漏洞后会给出PoC,之后根据指纹可以检测出更多的资产,此时针对多目标时就需要进行代码批量化检测,多数都需要使用到requests模块,主要是发送http/https请求,并获取响应结果,示例代码如下所示
import requests
x = requests.get('https://baidu.com')
print(x.content.decode('utf-8')) # 中文解析返回结果如下所示
<!DOCTYPE html>
<!--STATUS OK--><html> <head><meta http-equiv=content-type content=text/html;charset=utf-8><meta http-equiv=X-UA-Compatible content=IE=Edge><meta content=always name=referrer><link rel=stylesheet type=text/css href=https://ss1.bdstatic.com/5eN1bjq8AAUYm2zgoY3K/r/www/cache/bdorz/baidu.min.css><title>百度一下,你就知道</title></head> <body link=#0000cc> <div id=wrapper> <div id=head> <div class=head_wrapper> <div class=s_form> <div class=s_form_wrapper> <div id=lg> <img hidefocus=true src=//www.baidu.com/img/bd_logo1.png width=270 height=129> </div> <form id=form name=f action=//www.baidu.com/s class=fm> <input type=hidden name=bdorz_come value=1> <input type=hidden name=ie value=utf-8> <input type=hidden name=f value=8> <input type=hidden name=rsv_bp value=1> <input type=hidden name=rsv_idx value=1> <input type=hidden name=tn value=baidu><span class="bg s_ipt_wr"><input id=kw name=wd class=s_ipt value maxlength=255 autocomplete=off autofocus=autofocus></span><span class="bg s_btn_wr"><input type=submit id=su value=百度一下 class="bg s_btn" autofocus></span> </form> </div> </div> <div id=u1> <a href=http://news.baidu.com name=tj_trnews class=mnav>新闻</a> <a href=https://www.hao123.com name=tj_trhao123 class=mnav>hao123</a> <a href=http://map.baidu.com name=tj_trmap class=mnav>地图</a> <a href=http://v.baidu.com name=tj_trvideo class=mnav>视频</a> <a href=http://tieba.baidu.com name=tj_trtieba class=mnav>贴吧</a> <noscript> <a href=http://www.baidu.com/bdorz/login.gif?login&tpl=mn&u=http%3A%2F%2Fwww.baidu.com%2f%3fbdorz_come%3d1 name=tj_login class=lb>登录</a> </noscript> <script>document.write('<a href="http://www.baidu.com/bdorz/login.gif?login&tpl=mn&u='+ encodeURIComponent(window.location.href+ (window.location.search === "" ? "?" : "&")+ "bdorz_come=1")+ '" name="tj_login" class="lb">登录</a>');
</script> <a href=//www.baidu.com/more/ name=tj_briicon class=bri style="display: block;">更多产品</a> </div> </div> </div> <div id=ftCon> <div id=ftConw> <p id=lh> <a href=http://home.baidu.com>关于百度</a> <a href=http://ir.baidu.com>About Baidu</a> </p> <p id=cp>©2017 Baidu <a href=http://www.baidu.com/duty/>使用百度前必读</a> <a href=http://jianyi.baidu.com/ class=cp-feedback>意见反馈</a> 京ICP证030173号 <img src=//www.baidu.com/img/gs.gif> </p> </div> </div> </div> </body> </html>
每次调用requests请求之后,都会返回一个response对象,该对象包含了具体的响应信息,如状态码、响应头以及响应内容等,部分响应信息如下所示
apparent_encoding 编码方式
close() 关闭与服务器的连接
content 返回响应内容,以字节为单位
cookies 返回一个CookieJar对象,包含了从服务器发回的cookie
elapsed 返回一个timedelta对象,发送到响应的时间量,可用于测试响应速度
encoding 解码x.text的编码方式
headers 返回响应头,字典格式
history 返回包含请求历史的相应对象列表
is_permanent_redirect 如果响应是永久重定向的url,则返回true,反之false
is_redirect 如果响应被重定向,则返回true,反之false
iter_content() 迭代响应
iter_lines() 迭代响应的行
json() 返回结果的JSON对象(结果需要以JSON格式编写,否则出错)
links 返回响应的解析头链接
next 返回重定向链中下一个请求的PreparedRequest对象
raise_for_status() 如发生错误,返回一个HTTPError对象
reason 响应状态的描述
request 返回请求此响应的请求对象
status_code 返回http状态码
text 返回响应内容,unicode类型数据
url 返回响应的URL
对于requests的使用方法如下所示
delete(url,args) 发送DELETE请求到指定url
get(url,params,args) 发送GET请求到指定url
head(url,args) 发送HEAD请求到指定url
patch(url,data,args) 发送PATCH请求到指定url
post(url,data,json,args) 发送POST请求到指定url
put(url,data,args) 发送PUT请求到指定url
request(method,url,args) 向指定url发送指定的请求方法
在requests基本方法知道后,我们可以进一步的使用该模块,如果在撰写exp时,发现会对部分请求头进行检测时,我们可以自定义加入,使得请求更加完整,示列代码如下所示
import requests
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3'
}
resp = requests.get('https://httpbin.org/get', headers=headers)
print(resp.json()) # 返回格式为json上面的样例代码主要是针对的GET请求方式,下面是POST示例代码
import requests
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3'
}
payload = {
'param1': 'value1',
'param2': 'value2'
}
resp = requests.post('https://httpbin.org/post', headers=headers, data=payload)
print(resp.text) # 与上面的json区别输出如下所示,上面的GET代码也是类似,只是用json()格式化了
{
"args": {},
"data": "",
"files": {},
"form": {
"param1": "value1",
"param2": "value2"
},
"headers": {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate",
"Content-Length": "27",
"Content-Type": "application/x-www-form-urlencoded",
"Host": "httpbin.org",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3",
"X-Amzn-Trace-Id": "Root=1-696266c5-599454905262024b7c21fc2e"
},
"json": null,
"origin": "23.249.26.213",
"url": "https://httpbin.org/post"
}
写一个用于测试DVWA中的SQLI部分(下面是我在CCISC比赛时写的WAF批量自动化测试脚本中的测试函数,仅供参考)
def login_and_sqli_test(username,password,target,dvwa_url):
print('--'*40)
print(f"[+] 正在测试{target}--{dvwa_url}\n")
# 拿 user_token,等价 CSRF_Token,并进行持久化
session = requests.Session()
req_token = session.get(f'{dvwa_url}/login.php')
soup = BeautifulSoup(req_token.text,'html.parser')
user_token = soup.find('input',attrs={'name':'user_token'})['value']
headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36"
}
# 拿验证 Token
login_data = {
'username': username,
'password': password,
'Login': 'Login',
'user_token': user_token
}
req_login = session.post(f'{dvwa_url}/login.php',data=login_data)
# 设置为 Low 确保防护完全由 WAF 来
security_data = {
'security': 'low',
'seclev_submit': 'Submit',
'user_token': user_token
}
req_security = session.post(f'{dvwa_url}/security.php',data=security_data)
# SQLi 测试部分
for file_type in files:
match = re.search(r'/([^/]+)\.txt$',file_type)
test_type = match.group(1)
with open(file_type,'r') as f:
payloads = f.readlines()
payloads = [payload.strip() for payload in payloads]
print(f"[+] 开始测试 {test_type}")
for payload in payloads:
# print(f"[+] 正在测试 {payload} ")
sqli_data = {
'id': '1' + payload,
'Submit': 'Submit'
}
req_sqli = session.get(f'{dvwa_url}/vulnerabilities/sqli/',params=sqli_data)
# 统计数据
if req_sqli.status_code == 200:
result[target][test_type] += 1
print(f"[+] {test_type} 已测试完成\n")
time.sleep(2)
print(f"[+] {dvwa_url} 已测试完成\n")
print('--'*40)
return result通过查询接口实现(非API)来进行IP是否恶意的查询,其中会涉及到一个认证头的使用,这是认证头地址https://www.virustotal.com/gui/main.d3a4ab5b29d6a93642df.js
import re,requests,base64,random,time
# Virustotal computeAntiAbuseHeader接口
def computeAntiAbuseHeader():
e = time.time()
n = 1e10 * (1 + random.random() % 5e4)
raw = f'{n:.0f}-ZG9udCBiZSBldmls-{e:.3f}'
res = base64.b64encode(raw.encode())
return res.decode()
# 提IP
def extract_ips(file_path):
with open(file_path, "r", encoding='utf-8') as f:
content = f.read()
return re.findall(r"(?:\d{1,3}\.){3}\d{1,3}", content)
def query_ip(ip):
url = f"https://www.virustotal.com/ui/ip_addresses/{ip}"
headers = {
"Accept-Language": "en-US,en;q=0.9,es;q=0.8",
"sec-ch-ua-platform": "macOS",
"Referer": "https://www.virustotal.com/",
"sec-ch-ua": '"Chromium";v="140", "Not=A?Brand";v="24", "Microsoft Edge";v="140"',
"X-VT-Anti-Abuse-Header": computeAntiAbuseHeader(),
"sec-ch-ua-mobile": "?0",
"X-Tool": "vt-ui-main",
"x-app-version": "v1x460x0",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36",
"accept": "application/json",
"Content-Type": "application/json"
}
try:
resp = requests.get(url=url, headers=headers, timeout=10)
data = resp.json()
stats = data.get('data', {}).get('attributes', {}).get('last_analysis_stats', {})
return stats
except Exception as e:
print(f"[!] 查询 {ip} 出错: {e}")
return {}
def main():
file_path = input("请输入文件路径:").strip()
ips = extract_ips(file_path)
print(ips)
if not ips:
print(f"[!] 没有任何 IP 可提取")
return
for ip in ips:
stats = query_ip(ip)
if not stats:
continue
malicious = stats.get("malicious", 0)
vt_url = f"https://www.virustotal.com/ui/ip_addresses/{ip}"
if malicious > 0:
print(f"IP: {ip}\nVT地址: {vt_url}\n结果: {malicious}/94 security vendor flagged this IP address as malicious\n")
else:
print(f"IP: {ip} 为正常 IP\n")
if __name__ == "__main__":
main()Re模块主要是通过一系列正则匹配去查找特定的字符,这在安全领域中也是常用的,因为可以通过此进行安全检测,或是规则的模糊测试,下面是主要的正则表达式语法
1.常规单词
2.特殊字符:
.(匹配任意单个字符,换行符除外)
*(匹配前面的字符零次或多次)
+(匹配前面的字符一次或多次)
?(匹配前面的字符零次或多次)
\d(匹配任意数字字符)
\w(匹配任意字母、数字或下划线字符)
3.字符集
[abc]匹配a/b/c中任一字符
4.分组匹配
(ab)+ 表示匹配ab这种组合的内容
下面是核心函数
re.compile(pattern) 预编译正则表达式 pat = re.compile(r'\d+')
re.search(pattern,string) 搜索字符串中第一个匹配项
re.fullmatch(pattern,string) 整个字符串完全匹配
在re模块的match方法里面,主要是从字符串的起始位置开始匹配,其语法如下所示
re.match(pattern,string,flags=0)其中,pattern参数为将要匹配的内容/表达式,而string参数为要进行匹配的字符串,最后的flags为可选参数,主要有如下可选
re.I 使匹配对大小写不敏感
re.L 做本地化识别匹配
re.M 多行匹配,影响^与$
re.S 使.匹配包括换行在内的所有字符
re.U 根据Unicode字符集解析字符,影响\w \W \b \B
re.X 该标志通过给予你更灵活的格式以便你将正则表达式写得更易于理解
因此,如果匹配的内容不在起始位置的话,是不能匹配到的,示例代码如下所示
import re
text = "' union select 1,2,3#"
findchr = "select"
if re.match(pattern=findchr,string=text):
print("Match!")
else:
print("No Match")输出结果如下所示
No Match
与3.3.1的区别在于,该方法是从字符串中进行搜索,而不是在开头进行,示例代码如下所示
import re
finding = r'union'
text = "'/**/union/**/select+1,2,3#"
if re.search(pattern=finding,string=text):
print("Match")
else:
print("No Match")输出结果如下所示
Match
该方法主要是用于在字符串中按照特定的正则表达式进行匹配的子串,并返回一个包含这些子串的列表,示例代码如下所示
import re
regx = r'\d+'
text = "select 1,2,3#"
print(re.findall(pattern=regx,string=text))输出结果如下所示
['1', '2', '3']
该方法主要是用来替换符合匹配规则部分为其他内容的方法,示例代码如下所示
import re
regx = r'or'
text = "admin' or 1=1#"
print(re.sub(pattern=regx,string=text,repl="||"))输出结果如下所示
admin' || 1=1#