Import Feishu directory with optional 30-minute sync

This commit is contained in:
Codex committed 2026-10-23 11:10:59 +08:00
1 parent 4096b93da3
commit e5375deeb2
10 files changed
+662 -3

No files matched your search

+3
View File
@@ -315,6 +315,9 @@ func (s *Server) RegisterRoutes(r *gin.Engine) {
admin.GET("/system/identity", idp.Status)
admin.GET("/system/identity/profiles", idp.Profiles)
admin.POST("/system/identity/directory-check", idp.DirectoryCheck)
admin.GET("/system/identity/directory-sync", idp.DirectorySyncStatus)
admin.PUT("/system/identity/directory-sync", idp.DirectorySyncConfigure)
admin.POST("/system/identity/directory-sync/run", idp.DirectorySyncRun)
admin.GET("/tunnels/:id/feishu-grants", idp.Grants)
admin.POST("/tunnels/:id/feishu-grants", idp.AddGrant)
admin.DELETE("/tunnels/:id/feishu-grants/:grant_id", idp.DeleteGrant)
@@ -0,0 +1,414 @@
package identityprovider
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"sort"
"strings"
"sync"
"time"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"vpn-platform/backend/internal/model"
"vpn-platform/backend/internal/store"
)
const contactAPI = "https://open.feishu.cn/open-apis/contact/v3"
var directoryRunMu sync.Mutex
var directoryRunning bool
type directorySnapshot struct {
departments map[string]model.FeishuDepartment
members map[string]model.FeishuDirectoryMember
}
func (p *Provider) DirectorySyncStatus(c *gin.Context) {
noStore(c)
if !secureAdmin(c) {
c.JSON(403, gin.H{"error": "请通过可信 HTTPS 管理入口查看通讯录"})
return
}
cfg, base, err := enabled()
if err != nil {
c.JSON(409, gin.H{"error": "请先启用飞书身份服务"})
return
}
status := syncStatus(cfg, base)
var members []model.FeishuDirectoryMember
if err := store.DB.Where("issuer = ? AND app_id = ?", base, cfg.AppID).Order("name asc").Limit(500).Find(&members).Error; err != nil {
c.JSON(500, gin.H{"error": "读取成员失败"})
return
}
var departments []model.FeishuDepartment
if err := store.DB.Where("issuer = ? AND app_id = ?", base, cfg.AppID).Order("name asc").Limit(500).Find(&departments).Error; err != nil {
c.JSON(500, gin.H{"error": "读取部门失败"})
return
}
departmentItems := make([]gin.H, 0, len(departments))
for _, d := range departments {
departmentItems = append(departmentItems, gin.H{"id": d.Department, "name": d.Name, "parent_id": d.Parent, "last_seen_at": d.LastSeenAt})
}
directoryRunMu.Lock()
running := directoryRunning
directoryRunMu.Unlock()
c.JSON(200, gin.H{"status": status, "running": running, "members": members, "departments": departmentItems})
}
func syncStatus(cfg model.IdentityProvider, base string) model.FeishuDirectorySync {
var status model.FeishuDirectorySync
if store.DB.Where("id = ? AND issuer = ? AND app_id = ?", 1, base, cfg.AppID).First(&status).Error != nil {
status = model.FeishuDirectorySync{ID: 1, Issuer: base, AppID: cfg.AppID}
}
return status
}
func (p *Provider) DirectorySyncConfigure(c *gin.Context) {
noStore(c)
if !secureAdmin(c) {
c.JSON(403, gin.H{"error": "请通过可信 HTTPS 管理入口配置同步"})
return
}
cfg, base, err := enabled()
if err != nil {
c.JSON(409, gin.H{"error": "请先启用飞书身份服务"})
return
}
var input struct {
Enabled *bool `json:"enabled"`
}
if c.ShouldBindJSON(&input) != nil || input.Enabled == nil {
c.JSON(400, gin.H{"error": "配置格式错误"})
return
}
directoryRunMu.Lock()
defer directoryRunMu.Unlock()
status := syncStatus(cfg, base)
status.Enabled = *input.Enabled
if err := store.DB.Save(&status).Error; err != nil {
c.JSON(500, gin.H{"error": "保存同步设置失败"})
return
}
store.Audit(c.GetString("username"), "feishu_directory_schedule", cfg.AppID, fmt.Sprint(status.Enabled))
c.JSON(200, gin.H{"status": status})
}
func (p *Provider) DirectorySyncRun(c *gin.Context) {
noStore(c)
if !secureAdmin(c) {
c.JSON(403, gin.H{"error": "请通过可信 HTTPS 管理入口同步"})
return
}
directoryRunMu.Lock()
if directoryRunning {
directoryRunMu.Unlock()
c.JSON(409, gin.H{"error": "同步正在进行"})
return
}
directoryRunning = true
directoryRunMu.Unlock()
actor := c.GetString("username")
go func() {
defer func() { directoryRunMu.Lock(); directoryRunning = false; directoryRunMu.Unlock() }()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
status, err := performDirectorySync(ctx)
if err != nil {
store.Audit(actor, "feishu_directory_sync_failed", status.AppID, "")
} else {
store.Audit(actor, "feishu_directory_sync", status.AppID, fmt.Sprintf("departments=%d members=%d", status.Departments, status.Members))
}
}()
c.JSON(202, gin.H{"running": true})
}
var errDirectoryBusy = errors.New("directory sync busy")
func runDirectorySync(ctx context.Context) (model.FeishuDirectorySync, error) {
directoryRunMu.Lock()
if directoryRunning {
directoryRunMu.Unlock()
return model.FeishuDirectorySync{}, errDirectoryBusy
}
directoryRunning = true
directoryRunMu.Unlock()
defer func() { directoryRunMu.Lock(); directoryRunning = false; directoryRunMu.Unlock() }()
return performDirectorySync(ctx)
}
func performDirectorySync(ctx context.Context) (model.FeishuDirectorySync, error) {
cfg, base, err := enabled()
if err != nil {
return model.FeishuDirectorySync{}, errors.New("飞书身份服务未启用")
}
status := syncStatus(cfg, base)
now := time.Now()
status.LastAttempt = &now
status.LastError = ""
if err := store.DB.Save(&status).Error; err != nil {
return status, errors.New("保存同步状态失败")
}
secret, err := store.DecryptSecret(cfg.AppSecret)
if err != nil {
return failedSync(status, errors.New("读取飞书应用凭据失败"))
}
token, err := directoryTenantToken(ctx, cfg.AppID, secret)
if err != nil {
return failedSync(status, err)
}
snapshot, err := readDirectory(ctx, token)
if err != nil {
return failedSync(status, err)
}
now = time.Now()
status.LastSuccess, status.Departments, status.Members = &now, len(snapshot.departments), len(snapshot.members)
status.LastError = ""
err = store.DB.Transaction(func(tx *gorm.DB) error {
for _, dept := range snapshot.departments {
dept.Issuer, dept.AppID, dept.LastSeenAt = base, cfg.AppID, now
if err := tx.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "issuer"}, {Name: "app_id"}, {Name: "department"}}, DoUpdates: clause.AssignmentColumns([]string{"parent", "name", "last_seen_at"})}).Create(&dept).Error; err != nil {
return err
}
}
for _, member := range snapshot.members {
member.Issuer, member.AppID, member.LastSeenAt = base, cfg.AppID, now
if err := tx.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "issuer"}, {Name: "app_id"}, {Name: "open_id"}}, DoUpdates: clause.AssignmentColumns([]string{"name", "departments", "last_seen_at"})}).Create(&member).Error; err != nil {
return err
}
}
return tx.Save(&status).Error
})
if err != nil {
return failedSync(status, errors.New("写入通讯录失败"))
}
return status, nil
}
func failedSync(status model.FeishuDirectorySync, err error) (model.FeishuDirectorySync, error) {
status.LastError = err.Error()
_ = store.DB.Save(&status).Error
return status, err
}
func directoryTenantToken(ctx context.Context, appID, secret string) (string, error) {
payload, _ := json.Marshal(map[string]string{"app_id": appID, "app_secret": secret})
req, err := http.NewRequestWithContext(ctx, http.MethodPost, feishuTenantToken, strings.NewReader(string(payload)))
if err != nil {
return "", err
}
req.Header.Set("Content-Type", "application/json")
resp, err := upstreamClient.Do(req)
if err != nil {
return "", errors.New("连接飞书失败")
}
defer resp.Body.Close()
var result struct {
Code *int `json:"code"`
Token string `json:"tenant_access_token"`
}
if json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&result) != nil || resp.StatusCode != http.StatusOK || result.Code == nil || *result.Code != 0 || result.Token == "" {
return "", feishuFailure("获取应用凭据", resp.StatusCode, result.Code)
}
return result.Token, nil
}
func directoryGet(ctx context.Context, token, path string, query url.Values, out any) error {
u := contactAPI + path
if len(query) > 0 {
u += "?" + query.Encode()
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
if err != nil {
return err
}
req.Header.Set("Authorization", "Bearer "+token)
resp, err := upstreamClient.Do(req)
if err != nil {
return errors.New("飞书通讯录请求失败")
}
defer resp.Body.Close()
var result struct {
Code *int `json:"code"`
Data json.RawMessage `json:"data"`
}
if json.NewDecoder(io.LimitReader(resp.Body, 2<<20)).Decode(&result) != nil || resp.StatusCode != http.StatusOK || result.Code == nil || *result.Code != 0 {
return feishuFailure("读取飞书通讯录", resp.StatusCode, result.Code)
}
if json.Unmarshal(result.Data, out) != nil {
return errors.New("飞书通讯录返回格式错误")
}
return nil
}
func readDirectory(ctx context.Context, token string) (directorySnapshot, error) {
snapshot := directorySnapshot{departments: map[string]model.FeishuDepartment{}, members: map[string]model.FeishuDirectoryMember{}}
var scope struct {
DepartmentIDs []string `json:"department_ids"`
UserIDs []string `json:"user_ids"`
}
if err := directoryGet(ctx, token, "/scopes", url.Values{"department_id_type": {"open_department_id"}, "user_id_type": {"open_id"}}, &scope); err != nil {
return snapshot, err
}
if len(scope.DepartmentIDs) == 0 && len(scope.UserIDs) == 0 {
return snapshot, errors.New("飞书应用通讯录可见范围为空;请配置成员和部门可见范围")
}
queue, seen := append([]string{}, scope.DepartmentIDs...), map[string]bool{}
for len(queue) > 0 {
id := queue[0]
queue = queue[1:]
if id == "" || len(id) > 128 {
return snapshot, errors.New("飞书部门标识无效")
}
if seen[id] {
continue
}
seen[id] = true
if len(seen) > 500 {
return snapshot, errors.New("部门超过同步上限 500")
}
if _, ok := snapshot.departments[id]; !ok {
snapshot.departments[id] = model.FeishuDepartment{Department: id, Name: id}
}
if err := readDirectoryPages(ctx, token, "/departments/"+url.PathEscape(id)+"/children", url.Values{"department_id_type": {"open_department_id"}}, func(raw json.RawMessage) error {
var items []struct {
ID string `json:"open_department_id"`
Name string `json:"name"`
Parent string `json:"parent_department_id"`
}
if err := json.Unmarshal(raw, &items); err != nil {
return err
}
for _, d := range items {
if d.ID == "" || len(d.ID) > 128 || len(d.Name) > 256 {
return errors.New("飞书部门信息无效")
}
snapshot.departments[d.ID] = model.FeishuDepartment{Department: d.ID, Name: d.Name, Parent: d.Parent}
if !seen[d.ID] {
queue = append(queue, d.ID)
}
}
return nil
}); err != nil {
return snapshot, err
}
if err := readDirectoryPages(ctx, token, "/users", url.Values{"department_id": {id}, "department_id_type": {"open_department_id"}, "user_id_type": {"open_id"}}, func(raw json.RawMessage) error {
var items []struct {
OpenID string `json:"open_id"`
Name string `json:"name"`
Departments []string `json:"department_ids"`
}
if err := json.Unmarshal(raw, &items); err != nil {
return err
}
for _, u := range items {
if !strings.HasPrefix(u.OpenID, "ou_") || len(u.OpenID) > 128 || len(u.Name) > 256 {
return errors.New("飞书用户标识或姓名无效")
}
deps := u.Departments
if len(deps) == 0 {
deps = []string{id}
}
if old, ok := snapshot.members[u.OpenID]; ok {
var existing []string
_ = json.Unmarshal([]byte(old.Departments), &existing)
deps = append(deps, existing...)
}
unique := map[string]bool{}
for _, department := range deps {
unique[department] = true
}
deps = deps[:0]
for department := range unique {
deps = append(deps, department)
}
sort.Strings(deps)
encoded, _ := json.Marshal(deps)
snapshot.members[u.OpenID] = model.FeishuDirectoryMember{OpenID: u.OpenID, Name: u.Name, Departments: string(encoded)}
if len(snapshot.members) > 20000 {
return errors.New("成员超过同步上限 20000")
}
}
return nil
}); err != nil {
return snapshot, err
}
}
// Explicitly visible users outside visible departments must also be imported.
for _, id := range scope.UserIDs {
if _, ok := snapshot.members[id]; ok {
continue
}
if !strings.HasPrefix(id, "ou_") || len(id) > 128 {
return snapshot, errors.New("飞书成员标识无效")
}
name, err := fetchBasicName(ctx, token, id)
if err != nil {
return snapshot, errors.New("读取独立可见成员失败;请检查用户基础信息权限")
}
snapshot.members[id] = model.FeishuDirectoryMember{OpenID: id, Name: name, Departments: "[]"}
if len(snapshot.members) > 20000 {
return snapshot, errors.New("成员超过同步上限 20000")
}
}
return snapshot, nil
}
func readDirectoryPages(ctx context.Context, token, path string, query url.Values, consume func(json.RawMessage) error) error {
visited := map[string]bool{}
for page := 0; page < 300; page++ {
query.Set("page_size", "100")
var data struct {
Items json.RawMessage `json:"items"`
HasMore bool `json:"has_more"`
PageToken string `json:"page_token"`
}
if err := directoryGet(ctx, token, path, query, &data); err != nil {
return err
}
if err := consume(data.Items); err != nil {
return errors.New("飞书通讯录分页数据无效")
}
if !data.HasMore {
return nil
}
if data.PageToken == "" || visited[data.PageToken] {
return errors.New("飞书通讯录分页游标无效")
}
visited[data.PageToken] = true
query.Set("page_token", data.PageToken)
}
return errors.New("飞书通讯录分页超过上限")
}
// StartDirectoryScheduler never imports until an administrator explicitly enables it.
func StartDirectoryScheduler(ctx context.Context) {
go func() {
ticker := time.NewTicker(time.Minute)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
}
cfg, base, err := enabled()
if err != nil {
continue
}
status := syncStatus(cfg, base)
if !status.Enabled || (status.LastAttempt != nil && time.Since(*status.LastAttempt) < 30*time.Minute) {
continue
}
runCtx, cancel := context.WithTimeout(ctx, 5*time.Minute)
_, _ = runDirectorySync(runCtx)
cancel()
}
}()
}
@@ -0,0 +1,62 @@
package identityprovider
import (
"context"
"io"
"net/http"
"strings"
"testing"
)
func TestReadDirectoryTraversesDepartmentsAndPages(t *testing.T) {
previous := upstreamClient
t.Cleanup(func() { upstreamClient = previous })
calls := map[string]int{}
upstreamClient = &http.Client{Transport: transportFunc(func(req *http.Request) (*http.Response, error) {
if req.Header.Get("Authorization") != "Bearer test-token" {
t.Fatal("missing application token")
}
path := req.URL.Path
calls[path]++
body := `{"code":0,"data":{"items":[]}}`
switch path {
case "/open-apis/contact/v3/scopes":
body = `{"code":0,"data":{"department_ids":["0"],"user_ids":[]}}`
case "/open-apis/contact/v3/departments/0/children":
body = `{"code":0,"data":{"items":[{"open_department_id":"od_child","name":"研发","parent_department_id":"0"}]}}`
case "/open-apis/contact/v3/users":
if req.URL.Query().Get("department_id") == "0" && req.URL.Query().Get("page_token") == "" {
body = `{"code":0,"data":{"items":[{"open_id":"ou_1","name":"甲","department_ids":["0"]}],"has_more":true,"page_token":"next"}}`
} else if req.URL.Query().Get("department_id") == "0" && req.URL.Query().Get("page_token") == "next" {
body = `{"code":0,"data":{"items":[{"open_id":"ou_2","name":"乙"}]}}`
} else if req.URL.Query().Get("department_id") == "od_child" {
body = `{"code":0,"data":{"items":[{"open_id":"ou_1","name":"甲","department_ids":["od_child"]}]}}`
} else {
t.Fatalf("unexpected page: %s", req.URL)
}
case "/open-apis/contact/v3/departments/od_child/children":
default:
t.Fatalf("unexpected endpoint: %s", req.URL)
}
return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(body)), Header: make(http.Header)}, nil
})}
snapshot, err := readDirectory(context.Background(), "test-token")
if err != nil || len(snapshot.departments) != 2 || len(snapshot.members) != 2 || calls["/open-apis/contact/v3/users"] != 3 || snapshot.members["ou_1"].Departments != `["0","od_child"]` {
t.Fatalf("unexpected snapshot: %v, %#v, %#v", err, snapshot, calls)
}
}
func TestReadDirectoryFailsClosedOnIncompletePage(t *testing.T) {
previous := upstreamClient
t.Cleanup(func() { upstreamClient = previous })
upstreamClient = &http.Client{Transport: transportFunc(func(req *http.Request) (*http.Response, error) {
body := `{"code":0,"data":{"department_ids":["0"]}}`
if req.URL.Path != "/open-apis/contact/v3/scopes" {
body = `{"code":0,"data":{"items":[],"has_more":true}}`
}
return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(body)), Header: make(http.Header)}, nil
})}
if _, err := readDirectory(context.Background(), "token"); err == nil || !strings.Contains(err.Error(), "分页游标") {
t.Fatalf("accepted partial directory: %v", err)
}
}
+33
View File
@@ -39,6 +39,39 @@ type IdentityProfile struct {
UpdatedAt time.Time
}
// FeishuDirectorySync tracks a single application's import schedule and outcome.
type FeishuDirectorySync struct {
ID uint `gorm:"primaryKey"`
Enabled bool `json:"enabled"`
Issuer string `gorm:"size:256"`
AppID string `gorm:"size:128"`
LastAttempt *time.Time `json:"last_attempt"`
LastSuccess *time.Time `json:"last_success"`
LastError string `gorm:"size:512" json:"last_error"`
Departments int `json:"departments"`
Members int `json:"members"`
}
type FeishuDepartment struct {
ID uint `gorm:"primaryKey"`
Issuer string `gorm:"uniqueIndex:idx_feishu_department;size:256"`
AppID string `gorm:"uniqueIndex:idx_feishu_department;size:128"`
Department string `gorm:"uniqueIndex:idx_feishu_department;size:128"`
Parent string `gorm:"size:128"`
Name string `gorm:"size:256"`
LastSeenAt time.Time
}
type FeishuDirectoryMember struct {
ID uint `gorm:"primaryKey" json:"-"`
Issuer string `gorm:"uniqueIndex:idx_feishu_member;size:256" json:"-"`
AppID string `gorm:"uniqueIndex:idx_feishu_member;size:128" json:"-"`
OpenID string `gorm:"uniqueIndex:idx_feishu_member;size:128" json:"open_id"`
Name string `gorm:"size:256" json:"name"`
Departments string `gorm:"type:text" json:"department_ids"`
LastSeenAt time.Time `json:"last_seen_at"`
}
// FeishuTunnelGrant records explicit admission for a verified Feishu subject.
type FeishuTunnelGrant struct {
ID uint `gorm:"primaryKey" json:"id"`
+1 -1
View File
@@ -44,7 +44,7 @@ func Init(dataDir string) error {
}
// 兼容旧库:VPNUser 唯一约束曾是 username 单列全局唯一
db.Exec("DROP INDEX IF EXISTS idx_vpn_user")
if err := db.AutoMigrate(&model.User{}, &model.LDAPConfig{}, &model.MFADevice{}, &model.VPNUser{}, &model.Tunnel{}, &model.Node{}, &model.Plugin{}, &model.AuditLog{}, &model.VPNSession{}, &model.ConnectionLog{}, &model.TrafficPolicy{}, &model.TrafficSample{}, &model.ClientDevice{}, &model.ClientSession{}, &model.IdentityProvider{}, &model.IdentityGrant{}, &model.IdentityProfile{}, &model.FeishuTunnelGrant{}); err != nil {
if err := db.AutoMigrate(&model.User{}, &model.LDAPConfig{}, &model.MFADevice{}, &model.VPNUser{}, &model.Tunnel{}, &model.Node{}, &model.Plugin{}, &model.AuditLog{}, &model.VPNSession{}, &model.ConnectionLog{}, &model.TrafficPolicy{}, &model.TrafficSample{}, &model.ClientDevice{}, &model.ClientSession{}, &model.IdentityProvider{}, &model.IdentityGrant{}, &model.IdentityProfile{}, &model.FeishuTunnelGrant{}, &model.FeishuDirectorySync{}, &model.FeishuDepartment{}, &model.FeishuDirectoryMember{}); err != nil {
return err
}
DB = db
+3
View File
@@ -2,6 +2,7 @@
package main
import (
"context"
"embed"
"fmt"
"io/fs"
@@ -16,6 +17,7 @@ import (
"vpn-platform/backend/internal/api"
"vpn-platform/backend/internal/core"
"vpn-platform/backend/internal/identityprovider"
"vpn-platform/backend/internal/monitor"
"vpn-platform/backend/internal/service"
"vpn-platform/backend/internal/service/custom"
@@ -51,6 +53,7 @@ func main() {
}
return
}
identityprovider.StartDirectoryScheduler(context.Background())
// 注册内置协议
mgr := core.NewManager()