Python安全编程

socket.socket([family[, type[, proto]]])

#都是可选值，family 默认 AF_INET

服务端

s.bind()

#绑定地址到套接字，在 AF_INET 下时元组形式保存（host,port）

s.listen()

#TCP 监听。只有在转为监听后才能 accept

s.accept()

#被动接受

--------------------

客户端

s.connect()

#主动初始化 TCP 服务连接，一般格式是元组（hostname,port）

s.connect_ex()

#上一个的扩展，遇到错误不抛异常，而是返回出错码，没有错误返回 0

--------------------

公用

s.recv()

#接受 TCP 数据

s.send()

s.sendall()

s.recvfrom()

#接受 UDP 数据

s.sendto()

#发送 UDP 数据

s.close()

#关闭套接字

s.getpeername()

#返回连接的远程地址，元组（ipaddr,port）

s.getsockname()

#返回套接字自己的地址，元组（ipaddr,port）

import socket

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

host = socket.gethostname()

port = 9999

server.bind((host, port))

#设置最大连接数，超过后排队

server.listen(5)

while True:

    #建立客户端连接，conn 是生成的链接实例，这里只是等待链接

    conn,addr = serversocket.accept()      

	while Ture:

        data = conn.recv(1234) #接收数据

    	conn.send(data.upper()) #发送数据

    conn.close();

#客户端：

s=socket.socket()

s。connect((socket.gethostname,1111))

s.recv(1234)

s.close

s.setblocking(True/False)

#默认为 True，阻塞模式，send/recv/accept 都要等待返回才继续执行，非阻塞反之

#非阻塞中，若操作无法完成，会报错 socket.error

s.setsockopt(level,optname,value)

#设置给定套接字选项的值

s.getsockopt(level,optname[.buflen])

#返回套接字选项的值

ip=IP(src='192.168.0.1',dst='192.168.0.2')

send(ip,iface='ens37')

#参数用于指定端口，参数可选，此时 protocol 为 0，没有协议

sendp(Ether(dst)/ARP(hwsrc,psrc,hwdst,pdst)/'xx',iface)

#构造 ARP：

hwsrc	源MAC

psrc	源IP

hwdst	目标MAC

(p)dst	目标IP

'xx'	ARP报文（payload）

ans,unans=sr(IP(dst)/ICMP())#返回元组，其中是两个列表，Result 以及 Unanswered

#当指向不存在的同网段主机，ARP 广播找不到目标，需要手动终止

ans.show/summary/nsummary()

#查看。

ans,unans=srp(Ether(dst)/ARP(pdst),timeout=3,iface)

ans,uans=sr(IP()/TCP(sport,dport ,flags="S")) #S 即 SYN

#sport 可用 RandShort (),RandNum () 随机生成，可以两参指定范围

#有些时候不会写具体端口，而是对应网络协议的名字，效果一样

#fuzz (TCP (dport,flags)), 自动填充 sport

TCP：dport=[22,80,123]

for sent,rece in ans:

    if rece.haslayler(TCP) and str(rece[TCP].flags)=='SA':

        print(str(sent(TCP).dport)+"Open")

    if ...'RA':

        "Closed"

    if ...ICMP...'3':

        "Filterd"

for sent in uans:

    "Filterd"

#hasLayer，查找返回包中对应协议

#client 是 SSHClient 对象

sftp=client.open_sftp()

sftp.put('lcoal.txt','remote.txt')

import paramiko

ssh=paramiko.SSHClient()

ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())#自动保存主机名和密钥

ssh.connect(hostname='127.0.0.1',port=22,username='root',password='xxx')

stdin,stdout,stderr=ssh.exec_command('ls')

print(stdout.read().deccode('utf8'))

ssh.close()

#连接时使用突然 transport 封装

transport=paramiko.transport(("127.0.0.1",22))

transport.connect(username='root',password='xxx')

ssh=paramiko.SSHClient()

ssh._transport=transport

...

transport.close()

#登录使用密钥，也可以类似上面使用 transport 封装

private=paramiko.RSAKey.from_private_key_file(<path_to_key_file>)

...

ssh.connect(hostname='127.0.0.1',username="root",pkey=private)

...

from_transport(cls,t)#创建已经连通的 SFTP 客户端通道

put(localpath,remotepath,callback=None,confirm=True)

#confirm 会调用 stat () 方法查看文件状态，返回 ls -l 结果

get(remotepath,localpath,callback=None)

remove()

rename()

stat()#如果不想跟随符号链接，使用 lstat ()

listdir() #列出服务器目录下的文件，比起 stat (), 不能具体到文件，也不能得知类型

import paramiko

tran=paramiko.transport(('127.0.0.1',22))

tran.connect(username='root',password='xxx')

sftp=paramilo.SFTPClient.from_transport(tran)#这里使用通道，不是 ssh.open_sftp ()

sftp.put(localpath,remotepath)

sftp.get(remotepath,localpath)

tran.close()

import paramiko

class ssh(object):

    def __init__(self,hostdata):

        self.ip=hostdata['ip']

        self.port=hostdata['port']

        self.username=hostdata['user']

        self.passwd=hostdata['passwd']

        self.transport=paramiko.Transport((self.ip,self.port))

        self.transport.connect(username=self.username,password=self.passwd)

        self.ssh=paramiko.SSHClient()

        self.ssh._transport=self.transport

        self.sftp=self.ssh.open_sftp()

    def command(self,cmd):

        if (isinstance(cmd, (str, int, float))):

            stdin,stdout,stderr=self.ssh.exec_command(cmd)

            return stdout.read()

        else:

            self.resultlist=[]

            for c in cmd:

                stdin,stdout,stderr=self.ssh.exec_command(c)

                self.resultlist.append(stdout.read())

                return self.resultlist

    def get(self,remote_path,local_path):

        try:

            self.sftp.get(remote_path,local_path)

            return local_path

        except Exception as e:

            return print("ERROR: %s" % e)

    def put(self,local_path,remote_path):

        try:

            self.sftp.get(local_path,remote_path)

            return self.sftp.lstat(remote_path)

        except Exception as e:

            return print("ERROR: %s" % e)

    def close(self):

        self.sftp.close()

        self.transport.close()

if __name__ == '__main__':

    host_data={

        'ip':'127.0.0.1',

        'port':22,

        'user':"root",

        'passwd':'xxx',

    }

    host=ssh(host_data)

    print(host.command('ls -l').decode('utf-8'))

    print(host.get("./index.php",r"C:\"))

os.getcwd() //pwd

os.chdir()  //cd

os.listdir()

os.mkdir() //raise FileExistsError

os.rmdir()  //raise OSError ，如果目录不为空

os.remove()  //raise FileNotFoundError

os.rename(src,dst)

os.getenv()

os.system()

#other------

os.access(path,mode)//mode包括os.F_OK(存在)，os.R_OK/W_OK/X_OK(rwx),返回布尔值

os.chmod(path,mode)//import stat,mode自己搜有什么

os.chwon(path,uid,gid)

os.chroot(path) //改变当前进程目录

res=subprocess.run(['ls','-l'],capture_output=True,text=True)

#text 项自动处理编码，否则应传入字节流，并且输出也是字节流

#去掉 captrue_output 项会使得直接输出直接到终端

print(res.stdout)

subprocess.run(['grep','python'],input='hello\npython\nworld',text=True)

#相当于

grep 'python'

hello

python

world

#或者

echo -e "hello\npython\nworld" | grep 'python'

import subprocess

try:

	res=subprocess.run(['ls','NOT EXISTS FILE'],capture_output =True,text=True,check=True)

    #check 的作用是在失败 (非 0 退出码) 时强制抛出错误 CalledProcessError

except subprocess.CalledProcessError as e:

    print(f"Command Not Found:{e.stderr}{e.returncode}")

#启动子进程

process=subprocess.Popen(['ping','potatowo.top'],stdout=subprocess.PIPE,text=True)#这里将输出捕获到管道里，不直接输出

while True:

	out_put=process.stdout.readline()#逐行读取

    if out_put=="" and process.poll() is not None:

	#无输出，并且子进程已经结束

        break

    if out_put:

        print(output.strip())

return_code=process.poll()#poll 的返回值是退出码，未结束是 None，见上

printf(f"Process finished with retuen code {return_code}")

p1=subprocess.Popen(['ls','-l'],stdout=subprocess.PIPE)

p2=subprocess.Popen(['grep','py'],stdin=p1.stdout,stdout=subprocess.PIPE,text=True)

out_put=p2.communicate()[0]

#交互式输入输出：

#std_out,std_err=process.communicate(input="data")

wait() #阻塞

terminate() #发送终止信号（SIGTREM）

kill() #强制终止（SIGKILL）

#以下三个都是输入 / 输出流对象，使用特定方法写入 / 读取内容

stdin().write()

stdout().read()

stderr().read()

re.match(pattern,string,flags=0)

#从起始位置匹配一个模式，不在起始位置匹配成功也返回 None

re.serch(pattern,string,flags=0)

#返回第一个匹配到的

re.findall()

#查询所有匹配的子串，并返回一个列表

match=re.sub(pattern,"potaowo",text)

#替换匹配到的部分

if match:

    print(match.group)

#匹配电话号码

pattern=r"\d{3}-\d{4}-\d{4}"

text="My phone number is 123-456-789"

match=re.search(pattern,match.group)

if match:

	print(match.group)

#匹配邮箱

pattern=r"^[a-zA-Z0-9_.+-]+@[a-zA-z0-9]+\.[a-zA-Z0-9-.]+$"

email="example@example.com"

if re.match(pattern,email):

    print("is Email")

re.fullmatch(r'\d+','123') #整个字符串完全匹配

re.finditer(r'd+','alb2') #这个返回的是迭代器

re.split(r'\d+','a1b2c) #按照匹配项划分，匹配的部分删除，返回列表

group() #返回整个匹配的字符串

group(n) #返回第 n 个捕获组的内容

groups() #返回所有捕获组的元组

start/end() #匹配的起始 / 结束位置

span() #返回匹配范围

regex=re.compile(pattern,flags=0) #预编译正则表达式，生成 Pattern 对象

text="xxx"

result=regex.findall(text)

import json

#读取

with open('data.json','r') as file:

	data=json.load(file) #解析为类似结构

    print(data)

#写入，文件不存在会自动创建

with open('data.json','w') as file:

    data={

        "name" = "patatowo"

    }

    json.dump(data,file)

#这里要与 dumps 区分开来，dumps 是将对对应的结构转换成字符串，是类似于序列化的过程

   json_string=json.dumps(data)

name=json_data["name"]

address_street_1=json_data["address"]["street"][1]

#修改嵌套元素同理

import xml.etree.ElementTree as ET

xml_string = '<root><element>Some data</element></root>'

root=ET.fromstring(xml_string)

#从文件中读取

tree=ET.parse('example.xml')

root = tree.getroot()

title=root.find('title') #只找到具有该标签的第一个元素

book=root.findall('book')

price=book.attrib['price']

title_text=title.text

new_element=ET.Element('new')#创建新元素

new_sub_element=ET.SubElement(root,'new_sub')

#创建具有指定标签的子元素

root.move(title)

xml='''

<bokstore>

	<book>

		<title>POTATOWO</title>

		<author>POTATO</author>

		<price>$114514</price>

	</book>

	<book>

		<title>POTATOWO2</title>

		<author>POTATO2</author>

		<price>$514114</price>

	</book>

</bookstore>'''

root=ET.fromstring(xml_string)

for book in root.findall('book'):

    title=book.find('title').text

    author=book.find('author').text

    price=book.find('author').text

    print(f"result:{title}{author}{price}")

root=ET.Element('bookstore')

#构建 book1

book1=ET.SubElement(root,'book')

title1=ET.SubElement(book1,'title')

title1.text='POTATOWO'

author1=ET.SubElement(book1,'author')

price1=ET.SubElement(book1,'price')

price1.text='$114514' 

#book2 类似

...

#保存

tree=ET.ElementTree(root)

tree.write('books.xml')

#反过来解析

parse_tree=ET.parse('books.xml')

parse_root=parse_tree.getroot()

#下面类上

...

import base64

#编码

data=b'xxx'

encode=base64.b64encode(data) #只接受 bytes 类型的数据

print(encode.decode()) #上面的返回的也是 bytes，这里 decode 转回 str

#解码

data=b'xxx'

decode=base64.b64decode(data)

print(decode.decode())

base64.urlsafe_b64encode(data)

base64.urlsafe_b63ddecode(data)

data={"name"="potatowo","password"=base64.b64.encode(b'passwd')}

with open('data.json') as file:

	data=json.load(file)

    data=base64.b64encode(data["data"])

with open("decoded_image.png","wb") as img_out:

    img_out.write(base64.b64decode(encoded_img))

import hashlib

sha256_hash=hashlib.new('sha256')

sha256_hash.update(b'POTATOWO') #更新对应对象的内容

print(sha256_hash.hexdigest())#获取对应 16 进制表示的哈希值

md5_hash=hashlib.md5(b'POTATOWO')

print(md5_hash.digest())#获取对应的二进制表达式

import sys

print(sys.argv)

source = sys.argv[1]

import argparse

def main():

    parser=argparse.ArgumentParser(

        prog='666', #程序名

        description='test', #描述

        epilog='Copyright(r), 2025', #说明

    )

    #定义位置参数

    parser.add_argument('outfile')

    parser.add_argument('--host',default='localhost')

    parser.add_argument('--port',default='3306',type=int)

    parser.add_argument('-u',required=True,help='666')

    parser.add_argument('-g','--gz',action='store_true', required=False) #对于无需参数值的指定

#解析参数，其中 - h 参数是默认的，输出 help，无需手动设定。使用时无需错误处理，参数不对会自动退出

    args=parser.parse_args()

if __name__ == '__main__' :

    main()

import tkinter as tk

from tkinter import messagebox

def greet():

    name=entry.get()

    if name.strip():

        messagebox.showinfo("greet",f"Hello,{name}!")

    else:

        messagebox.showwarning("Input Error","Please enter your name")

root=tk.Tk()

root.title("Hello Tkinter App")

root.geometry("400x200")

label=tk.Label(root,text="Enter your name:",font=("Arial",12))#字体设置只能用元组

label.pack(pady=10)

entry=tk.Entry(root,width=30)

entry.pack(pady=5)

button=tk.Button(root,text="Say Hello",command=greet)#command 是控件函数，只传函数名

button.pack(pady=10)

root.mainloop() #进入消息循环，必须放在最后

#软件图标

root.iconbitmap('app.ico')

跨平台：

from PIL import Image,ImageTk

image=Image.open('ico.png')

photo=ImageTk.PhotoImage(image)

root.iconphoto(Flase,photo)

#窗口正中央

def center_window(window,width,height):

    screen_width=window.winfo_screenwidth()

    screen_height=window.winfo_screenheight()

    x=(screen_width//2) - (width//2)

    y=(screen_height//2) - (height//2)

    window.geometry(f"{width}x{height}+{x}+{y}")

center_window=(root,600,400)

#毛玻璃

root.attributes("-alpha",0.9)#90% 不透明度

#退出确认

def on_closing():

    if messagebox.askokcancel("退出确认"，"你确认要退出嘛？"):

        print("正在清理资源")

        root.destory()

root.protocol("WM_DELETE_WINDOW",on_closing)#拦截关闭事件

def safe_exit():

	response=messagebox.askyesorncancel(

        title="save?",

        message="save",

        icon="warning"

    )

    if response is True:

        save_file()

        root.destroy()

    elif response is False:

		root.destory()

    elif reponse is None:

		pass #取消关闭

text=tk.StringVar(value="defaut_value")

label=tk.Lable(root,textvariable=text)

text.set("new_value") #新值

btn= tk.Buttom(root,text="666:",command=lamnda: greet(“dd”))#显然这里是依赖注入

from fuctools import partial

btn= tk.Buttom(root,text="666:",command=partial(delete_item，item_id))

passwd=tk.Entry(root,show="*")#隐藏密码

vcmd=(root.register(only_numbers),"%S")

entry=tk.Entry(root,validate="key",validatecommand=vcmd)#限制输入为数字

text_area = tk.Text(undo=True,maxundo=50)#后悔药

scroll=tk.Scrollbar(root,orient="vertical",command=text_area.yview)#滑条

text_area.config(yscrollcommand=scroll.set)

text_widget.see(tk.END) #自动滚到底

main_area.pack(expand=True,fill="both")#fill，占据所有剩余空间，缩放时会自动调制大小

root.grid_columnconfigure(1,weight=1)

#权重也就是 0-1，这样也会随着缩放自动调整

left=tk.Frame(root)

left.pack(side="right",fill="y") 

right=tk.Frame(root)

right.pack(side="right",expand=True,fill='both')#占据剩余空间