devstar/services/devcontainer/devcontainer_utils.go

package devcontainer

import (
	"archive/tar"
	"context"
	"fmt"
	"io"
	"net"
	"net/url"
	"regexp"
	"strings"

	"code.gitea.io/gitea/models/db"
	"code.gitea.io/gitea/models/perm"
	"code.gitea.io/gitea/models/repo"
	"code.gitea.io/gitea/models/user"
	"code.gitea.io/gitea/modules/charset"
	"code.gitea.io/gitea/modules/git"
	"code.gitea.io/gitea/modules/log"
	"code.gitea.io/gitea/modules/setting"
	"code.gitea.io/gitea/modules/typesniffer"
	"code.gitea.io/gitea/modules/util"
	gitea_context "code.gitea.io/gitea/services/context"
	"github.com/google/uuid"
)

func IsAdmin(ctx context.Context, doer *user.User, repoID int64) (bool, error) {
	if doer.IsAdmin {
		return true, nil
	}

	e := db.GetEngine(ctx)
	teamMember, err := e.Table("team_user").
		Join("INNER", "team_repo", "`team_repo`.team_id = `team_user`.team_id").
		Join("INNER", "team_unit", "`team_unit`.team_id = `team_user`.team_id").
		Where("`team_repo`.repo_id = ? AND `team_unit`.access_mode = ? ",
			repoID, perm.AccessModeAdmin).
		And("team_user.uid = ?", doer.ID).Exist()
	if err != nil {
		return false, nil
	}
	if teamMember {
		return true, nil
	}
	return e.Get(&repo.Collaboration{RepoID: repoID, UserID: doer.ID, Mode: 3})
}

func GetFileContentByPath(ctx context.Context, repo *repo.Repository, path string) (string, error) {
	var err error
	cfg, err := setting.NewConfigProviderFromFile(setting.CustomConf)
	if err != nil {
		return "", err
	}
	// 1. 获取默认分支名
	branchName := repo.DefaultBranch
	if len(branchName) == 0 {
		branchName = cfg.Section("repository").Key("DEFAULT_BRANCH").Value()
	}

	// 2. 打开默认分支
	gitRepoEntity, err := git.OpenRepository(ctx, repo.RepoPath())
	if err != nil {
		return "", err
	}
	defer func(gitRepoEntity *git.Repository) {
		_ = gitRepoEntity.Close()
	}(gitRepoEntity)

	// 3. 获取分支名称
	commit, err := gitRepoEntity.GetBranchCommit(branchName)
	if err != nil {
		return "", err
	}

	entry, err := commit.GetTreeEntryByPath(path)
	if err != nil {
		return "", err
	}
	// No way to edit a directory online.
	if entry.IsDir() {

		return "", fmt.Errorf("%s entry.IsDir", path)
	}

	blob := entry.Blob()
	if blob.Size() >= setting.UI.MaxDisplayFileSize {

		return "", fmt.Errorf("%s blob.Size overflow", path)
	}

	dataRc, err := blob.DataAsync()
	if err != nil {

		return "", err
	}

	defer dataRc.Close()
	buf := make([]byte, 1024)
	n, _ := util.ReadAtMost(dataRc, buf)
	buf = buf[:n]

	// Only some file types are editable online as text.
	if !typesniffer.DetectContentType(buf).IsRepresentableAsText() {

		return "", fmt.Errorf("typesniffer.IsRepresentableAsText")
	}

	d, _ := io.ReadAll(dataRc)

	buf = append(buf, d...)
	if content, err := charset.ToUTF8(buf, charset.ConvertOpts{KeepBOM: true}); err != nil {
		log.Error("ToUTF8: %v", err)
		return string(buf), nil
	} else {
		return content, nil
	}

}

// FileExists returns true if a file exists in the given repo branch
func FileExists(path string, repo *gitea_context.Repository) (bool, error) {
	var branch = repo.BranchName
	if branch == "" {
		branch = repo.Repository.DefaultBranch
	}
	commit, err := repo.GitRepo.GetBranchCommit(branch)
	if err != nil {
		return false, err
	}
	if _, err := commit.GetTreeEntryByPath(path); err != nil {
		return false, err
	}
	return true, nil
}
func ParseImageName(imageName string) (registry, namespace, repo, tag string) {

	// 分离仓库地址和命名空间
	parts := strings.Split(imageName, "/")
	if len(parts) == 3 {
		registry = parts[0]
		namespace = parts[1]
		repo = parts[2]
	} else if len(parts) == 2 {
		registry = parts[0]
		repo = parts[1]
	} else {
		repo = imageName
	}
	// 分离标签
	parts = strings.Split(repo, ":")
	if len(parts) > 1 {
		tag = parts[1]
		repo = parts[0]
	} else {
		tag = "latest"
	}
	if registry == "" {
		registry = "docker.io"
	}
	return registry, namespace, repo, tag
}
func getSanitizedDevcontainerName(username, repoName string) string {
	regexpNonAlphaNum := regexp.MustCompile(`[^a-zA-Z0-9]`)
	sanitizedUsername := regexpNonAlphaNum.ReplaceAllString(username, "")
	sanitizedRepoName := regexpNonAlphaNum.ReplaceAllString(repoName, "")
	if len(sanitizedUsername) > 15 {
		sanitizedUsername = strings.ToLower(sanitizedUsername[:15])
	}
	if len(sanitizedRepoName) > 31 {
		sanitizedRepoName = strings.ToLower(sanitizedRepoName[:31])
	}
	newUUID, _ := uuid.NewUUID()
	uuidStr := newUUID.String()
	uuidStr = regexpNonAlphaNum.ReplaceAllString(uuidStr, "")[:15]
	return fmt.Sprintf("%s-%s-%s", sanitizedUsername, sanitizedRepoName, uuidStr)
}
func ReplacePortOfUrl(originalURL, targetPort string) (string, error) {
	// 解析原始 URL
	parsedURL, _ := url.Parse(originalURL)

	// 获取主机名和端口号
	host, _, err := net.SplitHostPort(parsedURL.Host)
	if err != nil {
		// 如果没有端口号，则 SplitHostPort 会返回错误
		// 这种情况下，Host 就是主机名
		host = parsedURL.Host
	}

	// 重新组装 Host 和端口
	parsedURL.Host = net.JoinHostPort(host, targetPort)

	// 生成新的 URL 字符串
	newURL := parsedURL.String()
	return newURL, nil
}
func GetPortFromURL(urlStr string) (string, error) {
	parsedURL, err := url.Parse(urlStr)
	if err != nil {
		return "", fmt.Errorf("解析URL失败: %v", err)
	}

	// 获取主机名和端口号
	_, port, err := net.SplitHostPort(parsedURL.Host)
	if err != nil {
		// 如果SplitHostPort失败，说明URL中没有明确指定端口
		// 需要根据协议判断默认端口
		switch parsedURL.Scheme {
		case "http":
			return "80", nil
		case "https":
			return "443", nil
		default:
			return "", fmt.Errorf("未知协议: %s", parsedURL.Scheme)
		}
	}

	// 如果端口存在，直接返回
	return port, nil
}

// addFileToTar 将文件添加到 tar 归档
func AddFileToTar(tw *tar.Writer, filename string, content string, mode int64) error {
	hdr := &tar.Header{
		Name: filename,
		Mode: mode,
		Size: int64(len(content)),
	}
	if err := tw.WriteHeader(hdr); err != nil {
		return err
	}
	if _, err := tw.Write([]byte(content)); err != nil {
		return err
	}
	return nil
}