PyJWT 和 python-jose 在处理JWT令牌处理的时候的差异和具体使用

import jwt
import datetime

# 创建一个JWT
payload = {
    "user_id": 123,
    "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
}
secret = 'your-secret-key'
token = jwt.encode(payload, secret, algorithm='HS256')

# 解码JWT
decoded = jwt.decode(token, secret, algorithms=['HS256'])
print(decoded)

from jose import jwt
from jose.exceptions import JWTError

# 创建一个JWT
payload = {
    "user_id": 123,
    "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
}
secret = 'your-secret-key'
token = jwt.encode(payload, secret, algorithm='HS256')

# 解码JWT
try:
    decoded = jwt.decode(token, secret, algorithms=['HS256'])
    print(decoded)
except JWTError as e:
    print(f"Token is invalid: {e}")

pip install python-jose

from jose import jwt, JWTError
from jose.exceptions import ExpiredSignatureError

# 定义密钥和有效负载
secret = 'your-secret-key'
payload = {
    "user_id": 123,
    "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
}

# 生成 JWT
token = jwt.encode(payload, secret, algorithm='HS256')

# 解码 JWT 并处理可能的异常
try:
    decoded = jwt.decode(token, secret, algorithms=['HS256'])
    print(decoded)
except ExpiredSignatureError:
    print("Token has expired")
except JWTError:
    print("Token is invalid")

from fastapi import FastAPI, Depends, HTTPException, status, Request
from fastapi.responses import JSONResponse
from jose import JWTError, jwt
from datetime import datetime, timedelta

app = FastAPI()

# 配置项，通常从配置文件中加载
JWT_SECRET_KEY = 'your_jwt_secret_key'
JWT_ISSUER = 'your_issuer'
JWT_AUDIENCE = 'your_audience'
JWT_EXPIRED_DAYS = 7
ALGORITHM = 'HS256'

def generate_token(user_info: dict, role_type: str):
    # 获取IP地址
    ip = user_info.get('ip', '')

    # 定义声明
    claims = {
        'id': user_info['id'],
        'email': user_info['email'],
        'name': user_info['name'],
        'nickname': user_info.get('nickname', ''),
        'phone_number': user_info.get('mobile_phone', ''),
        'gender': user_info.get('gender', ''),
        'full_name': user_info.get('full_name', ''),
        'company_id': user_info.get('company_id', ''),
        'company_name': user_info.get('company_name', ''),
        'dept_id': user_info.get('dept_id', ''),
        'dept_name': user_info.get('dept_name', ''),
        'role_type': role_type,
        'ip': ip,
        'mac_addr': '',  # 无法获得Mac地址
        'channel': ''
    }

    # 定义token过期时间
    expiration = datetime.utcnow() + timedelta(days=JWT_EXPIRED_DAYS)

    # 创建JWT token
    to_encode = {
        **claims,
        'iss': JWT_ISSUER,
        'aud': JWT_AUDIENCE,
        'exp': expiration
    }

    token = jwt.encode(to_encode, JWT_SECRET_KEY, algorithm=ALGORITHM)
    return token

@app.post('/token')
async def get_token(request: Request):
    # 模拟的用户信息
    user_info = {
        'id': 123,
        'email': '[email protected]',
        'name': 'John Doe',
        'nickname': 'Johnny',
        'mobile_phone': '123-456-7890',
        'gender': 'Male',
        'full_name': 'Johnathan Doe',
        'company_id': 'ABC123',
        'company_name': 'ABC Corp',
        'dept_id': 'Dept001',
        'dept_name': 'IT',
        'ip': request.client.host
    }
    role_type = 'Admin'

    token = generate_token(user_info, role_type)
    
    headers = {
        'access-token': token,
        'Authorization': f'Bearer {token}'
    }

    return JSONResponse(content={'token': token}, headers=headers)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

from fastapi import Depends, HTTPException, status
from jose import JWTError, jwt
from typing import Optional

JWT_SECRET_KEY = 'your_jwt_secret_key'
ALGORITHM = 'HS256'

def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("id")
        if user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Could not validate credentials",
                headers={"WWW-Authenticate": "Bearer"},
            )
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 允许匿名访问的路由
@app.get("/public")
def read_public_data():
    return {"message": "This is a public endpoint"}

# 需要授权的路由
@app.get("/protected")
def read_protected_data(current_user: dict = Depends(get_current_user)):
    return {"message": f"Hello, {current_user['name']}"}

# 模拟登录生成 JWT 令牌的路由
@app.post("/token")
def login():
    # 这里省略了身份验证过程，只是直接生成一个 JWT 令牌
    token = jwt.encode({"id": 1, "name": "John Doe"}, JWT_SECRET_KEY, algorithm=ALGORITHM)
    return {"access_token": token, "token_type": "bearer"}

    # JWT auth, required
    app.add_middleware(
        AuthenticationMiddleware,
        backend=JwtAuthMiddleware(),
        on_error=JwtAuthMiddleware.auth_exception_handler,
    )

to_encode = {
    "sub": "user_id",
    "aud": "your_audience",  # 设置aud声明
    "exp": datetime.utcnow() + timedelta(minutes=30)
}

token = jwt.encode(to_encode, settings.TOKEN_SECRET_KEY, algorithm=settings.TOKEN_ALGORITHM)

try:
    payload = jwt.decode(
        token,
        settings.TOKEN_SECRET_KEY,
        algorithms=[settings.TOKEN_ALGORITHM],
        audience="your_audience"  # 验证aud声明
    )
    print(f"Decoded payload: {payload}")
except JWTError as e:
    print(f"Token decode failed: {e}")

to_encode = {
    "sub": str(user_id),  # 确保 user_id 是字符串
    "aud": "your_audience",  # 其他字段
    "exp": datetime.utcnow() + timedelta(minutes=30)
}

token = jwt.encode(to_encode, settings.TOKEN_SECRET_KEY, algorithm=settings.TOKEN_ALGORITHM)

user_id = 123  # 假设 user_id 是一个整数
to_encode = {
    "sub": str(user_id),  # 将 user_id 转换为字符串
    "aud": "your_audience",
    "exp": datetime.utcnow() + timedelta(minutes=30)
}

token = jwt.encode(to_encode, settings.TOKEN_SECRET_KEY, algorithm=settings.TOKEN_ALGORITHM)

from pydantic import BaseModel

class User(BaseModel):
    username: str
    email: str
    password: str  # 敏感字段

user = User(username="user1", email="[email protected]", password="secret")

# 创建一个不包含敏感字段的字典
user_dict = user.dict(exclude={"password"})

print(user_dict)

from sqlalchemy import Column, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(String, primary_key=True)
    username = Column(String)
    email = Column(String)
    password = Column(String)  # 敏感字段

    def to_dict(self):
        # 移除敏感字段
        return {c.name: getattr(self, c.name) for c in self.__table__.columns if c.name != 'password'}

user = User(id="1", username="user1", email="[email protected]", password="secret")
print(user.to_dict())

class User:
    def __init__(self, username, email, password):
        self.username = username
        self.email = email
        self.password = password  # 敏感字段

    def to_dict(self):
        # 移除敏感字段
        return {
            "username": self.username,
            "email": self.email
        }

user = User(username="user1", email="[email protected]", password="secret")
print(user.to_dict())