Harbor 间通过API实现镜像复制

news2025/6/19 8:30:47

背景：

通过harbor搭建私有镜像仓库时，往往会设计出研发镜像仓库--》测试镜像仓库--》产品镜像仓库，甚至更多。那个多镜像间的复制将成了一个必须解决的问题。当然可以通过harbor自带的界面操作即可实现需求，也可以通过harbor提供的API来实现。本文将针对通过jenkins来实现不同harbor间的镜像复制。

材料：

Harbor 镜像仓库A，Harbor镜像仓库B，jenkins(可选)

实现：

1、贴脚本：

#!/usr/bin/python
# -*- coding: UTF-8 -*-

import requests
import json
import argparse
import re

# 构造函数传参
parser = argparse.ArgumentParser()
parser.add_argument('--url',type=str,default='http://harbor.com:9090')
parser.add_argument('--user',type=str,default='admin')
parser.add_argument('--pwd',type=str,default='Harbor12345')
parser.add_argument('--taskname',type=str,default='')

#获取harbor登录token
def get_user_token(url):

  user_agent = 'Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Mobile Safari/537.36'  

  headers = {"Content-Type":"application/json","Cookie":"sid=a2f1c8cffed832cb42e0d718b6339acb","user-agent":user_agent}

  response = requests.get(url+"/c/log_out", headers = headers)

  cookiejar = response.cookies

  cookiedict = requests.utils.dict_from_cookiejar(cookiejar)
  #获得请求结果中的headers信息
  headers = response.headers
  #获得请求结果中的token
  csrf_token = headers.get('X-Harbor-Csrf-Token')
  #X-Request-Id
  requestId = headers.get('X-Request-Id')
  # 获得csrf
  csrf = cookiedict.get("_gorilla_csrf")
  #整合headers
  headers = {"Cookie":"_gorilla_csrf="+csrf,"X-Harbor-Csrf-Token":csrf_token,"Content-Type":"application/json;charset=UTF-8","user-agent":user_agent}
  #print headers
  return headers;

# 初始化用户登录信息
def login_init_user_tokken(url,username,password):
  headers = get_user_token(url)
  params = {"principal":username,"password":password}
  response = requests.post(url+"/c/login", headers = headers,params = params)
  cookiejar = response.cookies
  cookiedict = requests.utils.dict_from_cookiejar(cookiejar)
  sid = cookiedict.get("sid")
  new_headers = {"Cookie":headers['Cookie']+";sid="+sid,"X-Harbor-Csrf-Token":headers['X-Harbor-Csrf-Token'],"Content-Type":"application/json;charset=UTF-8","user-agent":headers['user-agent']}
  return new_headers

# 执行指定的Harbor任务
def run_copy_task_list(url,headers,taskid):
  #初始化登录
  datas= {"policy_id": taskid}
  response = requests.post(url+"/api/v2.0/replication/executions",json = datas,headers=headers)
  returnstr = json.dumps(response.text)
  return response.status_code


#获取Harbor任务复制列表
def get_copy_task_list(url,username,password,taskname):
  #初始化登录
  headers = login_init_user_tokken(url,username,password)
  #print headers1
  response = requests.get(url+"/api/v2.0/replication/policies?page=1&page_size=10&name="+taskname,headers=headers)
  returndata = response.json()
  sid =returndata[0].get("id")
  print run_copy_task_list(url,headers,int(sid))

# main
def main():
  args = parser.parse_args()

  print get_copy_task_list(str(args.url),str(args.user),str(args.pwd),str(args.taskname))

main()

2、调用脚本：python replicatesManager.py --taskname="你的镜像复制任务名" --url="你的镜像地址" --user="admin" --pwd="Harbor12345" 。注：可通过脚本中默认值来减少调用时变量。

3、解说：harbor API的调用官方提供了2中方式进行鉴权，本脚本采用了其中第二种。鉴权思路：a、通过log_out 接口获得Cookie,X-Harbor-Csrf-Token 这2个变量。