测试开发笔记

K8S相关

使用阿里云服务器部署traefik,关联NLB负载均衡器，添加cert-manager免费证书认证

By 李镇伟 2024年9月19日

背景需求

我们有一个常规的网站服务需要部署，并且对外提供https访问，从经济的角度考虑，建议购买阿里云的【容器集群ACK】【网站域名】【NBL负载均衡】【共享带宽包】【云服务器ECS】，如果有文件需求还可以购买【对象存储】【NAS文件系统】等，有静态文件加速需求还可以购买CDN服务，本篇文章我使用最低需求(钱)，仅购买3台服务器，组成k8s集群，部署web网站，自动使用acme.sh申请证书，使用外部负载均衡器来打造一个最低限度的高可用生产环境。

忽略的细节

从阿里云官网购买【容器集群ACK】【网站域名】【NBL负载均衡】【共享带宽包】【云服务器ECS】本文忽略，默认读者已购买并添加好，并复制kubeconfig文件到服务器上，kubectl和helm程序已经安装好，接下来直接敲命令

1.安装cert-manager

helm repo add jetstack https://charts.jetstack.io
 
helm repo update
 
helm install \
  cert-manager jetstack/cert-manager \
  --namespace cert-manager \
  --create-namespace \
  --version v1.15.3 \
  --set installCRDs=true

2.配置cert-manager所需要的阿里云dns挑战

安装阿里云dns挑战所对应的webhook

helm repo add cert-manager-alidns-webhook https://devmachine-fr.github.io/cert-manager-alidns-webhook
helm repo update
helm install alidns-webhook cert-manager-alidns-webhook/alidns-webhook

添加alidns-secret.yaml 文件，注意这里的access-key和secret-key是要通过阿里云的accesskey功能去获取的，获取之后，通过echo命令获取base64加密后的文本，填入yaml文件中

echo -n "原始密钥" | base64 

# alidns-secret.yaml 文件
apiVersion: v1
kind: Secret
metadata:
  name: alidns-secret
  namespace: cert-manager
data:
  access-key: base64加密后的
  secret-key: base64加密后的

kubectl apply -f alidns-secret.yaml

添加letsencrypt-staging.yaml文件，注意不要修改groupName，因为我上面helm install alidns-webhook的时候使用的默认参数里的groupName是example.com，一定要修改groupName的话，需要两边同步修改

apiVersion: cert-manager.io/v1
kind: ClusterIssuer
metadata:
  name: letsencrypt-staging
spec:
  acme:
    email: 357244849@qq.com
    server: https://acme-v02.api.letsencrypt.org/directory
    privateKeySecretRef:
      name: letsencrypt-staging-account-key
    solvers:
    - dns01:
        webhook:
          groupName: example.com
          solverName: alidns-solver
          config:
            region: ""
            accessKeySecretRef:
              name: alidns-secret
              key: access-key
            secretKeySecretRef:
              name: alidns-secret
              key: secret-key

kubectl apply -f letsencrypt-staging.yaml

添加Certificate.yaml文件,这一步完成后，手动去阿里云的域名解析里添加对应的cname解析了，记录值填负载均衡器给到的域名，一般是nlb-xxxx.地域.nlb.aliyuncs.com

apiVersion: cert-manager.io/v1
kind: Certificate
metadata:
  name: hello-com
  namespace: traefik
spec:
  # The secretName will store certificate content
  secretName: hello-com-tls
  dnsNames:

  - "*.hello.com"
  - "ops.hello.com"
  issuerRef:
    name: letsencrypt-staging
    kind: ClusterIssuer

kubectl apply -f Certificate.yaml

3.安装traefik并关联对应的证书和负载均衡器

1.编辑一个values-traefik.yaml文件，可以参考我下面的配置，

providers:
  kubernetesCRD:
    allowCrossNamespace: true
  kubernetesIngress:
    publishedService:
      enabled: true # 让 Ingress 的外部 IP 地址状态显示为 Traefik 的 LB IP 地址
service:
  enabled: true
  loadBalancerClass: alibabacloud.com/nlb
  annotations:
    service.beta.kubernetes.io/alibaba-cloud-loadbalancer-id: "NLB的ID" # 关联阿里云NLB负载均衡器的ID。
    service.beta.kubernetes.io/alibaba-cloud-loadbalancer-force-override-listeners: "true"
  spec:
    externalTrafficPolicy: Local
# 这里不加的话，80和 443 会报没有权限
securityContext:
  capabilities:
    add:
      - NET_BIND_SERVICE
  runAsNonRoot: false
  runAsUser: 0
updateStrategy:
  # -- Customize updateStrategy: RollingUpdate or OnDelete
  type: RollingUpdate
  rollingUpdate:
    maxUnavailable: 1
    maxSurge: 0
ports:
  web:
    port: 80
    expose:
      default: true
    exposedPort: 80 # 对外的 HTTP 端口号，使用标准端口号在国内需备案
    redirectTo:
      port: websecure
  websecure:
    port: 443
    expose:
      default: true
    exposedPort: 443 # 对外的 HTTPS 端口号，使用标准端口号在国内需备案
logs:
  access:
     enabled: true
deployment:
  enabled: true
  replicas: 3
ingressRoute:
  dashboard:
    enabled: true
    matchRule: Host(`traefik.hello.com`) && (PathPrefix(`/dashboard`) || PathPrefix(`/api`))
    entryPoints: ["websecure"]
    middlewares:
      - name: traefik-dashboard-auth
extraObjects:
  - apiVersion: v1
    kind: Secret
    metadata:
      name: traefik-dashboard-auth-secret
    type: kubernetes.io/basic-auth
    stringData:
      username: hello
      password: thankyou

  - apiVersion: traefik.io/v1alpha1
    kind: Middleware
    metadata:
      name: traefik-dashboard-auth
    spec:
      basicAuth:
        secret: traefik-dashboard-auth-secret
# 关联cert-manager设置的秘钥
tlsStore:
  default:
    defaultCertificate:
      secretName: hello-com-tls

helm repo add traefik https://helm.traefik.io/traefik
helm repo update
helm upgrade --install traefik -n traefik -f values-traefik.yaml traefik/traefik

安装完成后，使用kubectl get svc -n traefik就能看到生成的loadbalancer了，通过阿里云控制台也可以看到网络型负载均衡器里面自动创建了对应的监听和服务器组，如需验证部署后的效果，可以用浏览器访问https://traefik.hello.com/dashboard 进行测试

python相关

python实现的一个简易注册机，用于离线工控机验证注册码

By 李镇伟 2024年9月3日

简易的设计流程

有一批工控机长期是断网离线的，而我们又想检查这批工控机的设备上的注册码是否已经过期，那该怎么办呢？工控机是交付出去了，需要用到非对称加密机制来设计，防止算法被破解之后，别人有能力能破解我们所有的工控机。下面废话不多说，直接上代码

服务端程序

1. 安装 `cryptography` 库

pip install cryptography

2. 生成密钥对（一次性操作）

在服务器端生成 RSA 公钥和私钥，这个过程只需要进行一次，并且私钥需要保密。

# 注册机密钥对生成代码
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization

# 生成 RSA 密钥对
private_key = rsa.generate_private_key(
    public_exponent=65537,
    key_size=2048,
)

# 将私钥保存到文件
with open("private_key.pem", "wb") as private_file:
    private_file.write(
        private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        )
    )

# 将公钥保存到文件
public_key = private_key.public_key()
with open("public_key.pem", "wb") as public_file:
    public_file.write(
        public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo,
        )
    )

private_key.pem 文件将用于服务器端签名注册码。（千万不要泄漏）

public_key.pem 文件可以安全地分发给客户端，用于验证签名。

3. 服务器端生成注册码并签名

服务器端生成注册码并用私钥签名。这里我们基于 IP、MAC 地址和注册时间生成注册信息。

# 注册码生成程序
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import serialization
import hashlib

# 加载私钥
with open("private_key.pem", "rb") as private_file:
    private_key = serialization.load_pem_private_key(
        private_file.read(),
        password=None,
    )


# 生成注册信息
def generate_registration_info(ip, mac, reg_time):
    return f"{ip}#{mac}#{reg_time}"


# 生成签名
def sign_registration_info(private_key, registration_info):
    # 先对注册信息进行哈希处理
    hash_value = hashlib.sha256(registration_info.encode()).digest()
    # 使用私钥对哈希值签名
    signature = private_key.sign(
        hash_value,
        padding.PSS(
            mgf=padding.MGF1(hashes.SHA256()),
            salt_length=padding.PSS.MAX_LENGTH,
        ),
        hashes.SHA256(),
    )
    return signature


# 示例使用
ip = "192.168.1.1"
mac = "00:1A:2B:3C:4D:5E"
reg_time = "2023-09-03T12:00:00"
registration_info = generate_registration_info(ip, mac, reg_time)
signature = sign_registration_info(private_key, registration_info)

print(f"注册信息: {registration_info}")
print(f"签名: {signature.hex()}")

客户端程序（可参考修改）

1.安装在工控机上，用于验证注册码是否正确，是否过期，设置有效期365天

2.客户端使用公钥来验证签名，以确保注册码的有效性。

# 注册机尝试
import hashlib
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import serialization
from datetime import datetime

# 加载公钥
with open("public_key.pem", "rb") as public_file:
    public_key = serialization.load_pem_public_key(public_file.read())


# 验证签名
def verify_registration_info(public_key, registration_info, signature):
    # 对注册信息进行哈希处理
    hash_value = hashlib.sha256(registration_info.encode()).digest()
    # 使用公钥验证签名
    try:
        public_key.verify(
            signature,
            hash_value,
            padding.PSS(
                mgf=padding.MGF1(hashes.SHA256()),
                salt_length=padding.PSS.MAX_LENGTH,
            ),
            hashes.SHA256(),
        )
        return True
    except Exception as e:
        print(f"签名验证失败: {e}")
        return False


# 检查注册码是否过期
def is_registration_expired(registration_info):
    # 假设 registration_info 的格式为 "IP#MAC#DATE"
    try:
        _, _, date_str = registration_info.split("#")
        registration_date = datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S")
        current_date = datetime.now()
        # 计算日期差
        difference = current_date - registration_date
        if difference.days > 365:
            print("注册码已过期。")
            return True
        else:
            print(f"注册码有效，距离到期还有 {365 - difference.days} 天。")
            return False
    except Exception as e:
        print(f"无法解析注册信息中的日期: {e}")
        return True


# 示例使用
registration_info = "192.168.1.1#00:1A:2B:3C:4D:5E#2023-09-03T12:00:00"
signature = "4747554cb3bc12f1b25f7079f435338f78f8b9a096f7a40fe6f4d22535d5d1038ae0cff10f1f7d648201e585b19b15d45a0b0903886abda9096f7ce3dbba78b3076f7df9cec9f19616512c7dd20a4448ab1eb544be0163e84cb811cf415986551f4be21878c4ca620843e4109d4e625756560f0746dbbccd2f57ee1ba6d8f751bde45839e75229160371c955bc8d19931d647d1281f4ae6baa08dd460dbf3de0d1ac76f2217ed8e81657cce00da6342ef5d453afb0c24da10197896f89347a3dc81a482eab2e41ffe311222e86d0e6a0901ae6cce38e69700d7d4a18c9e902ea802b05292c166d561b4877619283026542e319ca35708fa32e6f86e5615f6fb0"
is_valid = verify_registration_info(
    public_key, registration_info, bytes.fromhex(signature)
)

if is_valid:
    print("注册码验证成功。")
    # 检查注册码是否过期
    is_expired = is_registration_expired(registration_info)
    if not is_expired:
        print("激活成功！")
else:
    print("注册码验证失败。")

月度归档： 2024 年 9 月