pub源代码:
https://github.com/dart-lang/pub

pub publish 核心代码：

OAuth2认证

   try {
      final officialPubServers = {
        'https://pub.dev',
        // [validateAndNormalizeHostedUrl] normalizes https://pub.dartlang.org
        // to https://pub.dev, so we don't need to do allow that here.

        // Pub uses oauth2 credentials only for authenticating official pub
        // servers for security purposes (to not expose pub.dev access token to
        // 3rd party servers).
        // For testing publish command we're using mock servers hosted on
        // localhost address which is not a known pub server address. So we
        // explicitly have to define mock servers as official server to test
        // publish command with oauth2 credentials.
        if (runningFromTest &&
            Platform.environment.containsKey('_PUB_TEST_DEFAULT_HOSTED_URL'))
          Platform.environment['_PUB_TEST_DEFAULT_HOSTED_URL'],
      };

      // Using OAuth2 authentication client for the official pub servers
      final isOfficialServer = officialPubServers.contains(host.toString());
      if (isOfficialServer && !cache.tokenStore.hasCredential(host)) {
        // Using OAuth2 authentication client for the official pub servers, when
        // we don't have an explicit token from [TokenStore] to use instead.
        //
        // This allows us to use `dart pub token add` to inject a token for use
        // with the official servers.
        await oauth2.withClient(cache, (client) {
          return _publishUsingClient(packageBytes, client);
        });
      } else {
        // For third party servers using bearer authentication client
        await withAuthenticatedClient(cache, host, (client) {
          return _publishUsingClient(packageBytes, client);
        });
      }
    } on PubHttpResponseException catch (error) {
      var url = error.response.request!.url;
      if (Uri.parse(url.origin) == Uri.parse(host.origin)) {
        handleJsonError(error.response);
      } else {
        rethrow;
      }
    }

上述实现流程如下：

定义了一个 officialPubServers 变量，表示所有官方的 Pub 仓库地址。

使用 OAuth2 认证客户端连接官方的 Pub 仓库，如果不存在 OAuth2 认证信息则通过命令行工具添加。

如果不是官方的 Pub 仓库则使用 bearer 认证客户端连接。

调用 _publishUsingClient 函数来发布包。

如果发布失败，并且失败的请求的地址是 Pub 仓库地址，则通过 handleJsonError 函数处理错误。如果不是 Pub 仓库地址，则重新抛出异常。

发布包

 try {
      await log.progress('Uploading', () async {
        /// 1. Initiate upload
        final parametersResponse =
            await retryForHttp('initiating upload', () async {
          final request =
              http.Request('GET', host.resolve('api/packages/versions/new'));
          request.attachPubApiHeaders();
          request.attachMetadataHeaders();
          return await client.fetch(request);
        });
        final parameters = parseJsonResponse(parametersResponse);

        /// 2. Upload package
        var url = _expectField(parameters, 'url', parametersResponse);
        if (url is! String) invalidServerResponse(parametersResponse);
        cloudStorageUrl = Uri.parse(url);
        final uploadResponse =
            await retryForHttp('uploading package', () async {
          // TODO(nweiz): Cloud Storage can provide an XML-formatted error. We
          // should report that error and exit.
          var request = http.MultipartRequest('POST', cloudStorageUrl!);

          var fields = _expectField(parameters, 'fields', parametersResponse);
          if (fields is! Map) invalidServerResponse(parametersResponse);
          fields.forEach((key, value) {
            if (value is! String) invalidServerResponse(parametersResponse);
            request.fields[key] = value;
          });

          request.followRedirects = false;
          request.files.add(
            http.MultipartFile.fromBytes(
              'file',
              packageBytes,
              filename: 'package.tar.gz',
            ),
          );
          return await client.fetch(request);
        });

        /// 3. Finalize publish
        var location = uploadResponse.headers['location'];
        if (location == null) throw PubHttpResponseException(uploadResponse);
        final finalizeResponse =
            await retryForHttp('finalizing publish', () async {
          final request = http.Request('GET', Uri.parse(location));
          request.attachPubApiHeaders();
          request.attachMetadataHeaders();
          return await client.fetch(request);
        });
        handleJsonSuccess(finalizeResponse);
      });
    } on AuthenticationException catch (error) {
      var msg = '';
      if (error.statusCode == 401) {
        msg += '$host package repository requested authentication!\n'
            'You can provide credentials using:\n'
            '    $topLevelProgram pub token add $host\n';
      }
      if (error.statusCode == 403) {
        msg += 'Insufficient permissions to the resource at the $host '
            'package repository.\nYou can modify credentials using:\n'
            '    $topLevelProgram pub token add $host\n';
      }
      if (error.serverMessage != null) {
        msg += '\n${error.serverMessage!}\n';
      }
      dataError(msg + log.red('Authentication failed!'));
    }

上述代码流程：

• Initiate upload：发送一个 GET 请求到 api/packages/versions/new 这个接口，以获取上传包的一些参数。请求会带有 pubApiHeaders 和 metadataHeaders 两个请求头，具体信息根据代码未知；

• Upload package：使用获取的参数，通过发送一个带有多个部分的 POST 请求，将包文件上传到云存储服务。在请求中，会有一些字段，具体内容在 fields 字段中，一般用于描述文件的一些元数据；

• Finalize publish：根据第二步的响应结果中的 location 字段，发送一个 GET 请求以确认发布完成。请求同样带有 pubApiHeaders 和 metadataHeaders 两个请求头。

上面的代码中的 retryForHttp 是一个重试机制，在某些情况下，请求失败时会重试多次，以获得成功的请求。

最后，handleJsonSuccess 方法被调用，该方法可以判断最终的响应结果是否是有效的 JSON 格式，如果不是，会抛出异常。

第二步请求地址：
https://storage.googleapis.com

返回第三步请求地址：

https://pub.dev/api/packages/versions/newUploadFinish?
upload_id=1c23e23c-c08d-4c08-bde4-b40bc58fe3b8
&bucket=dartlang-pub-incoming-packages
&key=tmp/1c23e23c-c08d-4c08-bde4-b40bc58fe3b8&etag=1846afb2360418d23d6c17db2bf90c6d

如果账号权限不一致 则在第三步返回：
xxx@gmail.com has insufficient permissions to upload new versions to existing package http.

如果是版本已经存在第三步返回：
Version 0.13.27 of package httptest already exists.